Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 299 insertions, 0 deletions

diff --git a/src/seastar/demos/rpc_demo.cc b/src/seastar/demos/rpc_demo.cc
new file mode 100644
index 000000000..d46cd07ec
--- /dev/null
+++ b/src/seastar/demos/rpc_demo.cc
@@ -0,0 +1,299 @@
+/*
+ * 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 2015 Cloudius Systems
+ */
+#include <cmath>
+#include <seastar/core/reactor.hh>
+#include <seastar/core/app-template.hh>
+#include <seastar/rpc/rpc.hh>
+#include <seastar/core/sleep.hh>
+#include <seastar/rpc/lz4_compressor.hh>
+#include <seastar/util/log.hh>
+#include <seastar/core/loop.hh>
+
+using namespace seastar;
+
+struct serializer {
+};
+
+template <typename T, typename Output>
+inline
+void write_arithmetic_type(Output& out, T v) {
+    static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
+    return out.write(reinterpret_cast<const char*>(&v), sizeof(T));
+}
+
+template <typename T, typename Input>
+inline
+T read_arithmetic_type(Input& in) {
+    static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
+    T v;
+    in.read(reinterpret_cast<char*>(&v), sizeof(T));
+    return v;
+}
+
+template <typename Output>
+inline void write(serializer, Output& output, int32_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, uint32_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, int64_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, uint64_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, double v) { return write_arithmetic_type(output, v); }
+template <typename Input>
+inline int32_t read(serializer, Input& input, rpc::type<int32_t>) { return read_arithmetic_type<int32_t>(input); }
+template <typename Input>
+inline uint32_t read(serializer, Input& input, rpc::type<uint32_t>) { return read_arithmetic_type<uint32_t>(input); }
+template <typename Input>
+inline uint64_t read(serializer, Input& input, rpc::type<uint64_t>) { return read_arithmetic_type<uint64_t>(input); }
+template <typename Input>
+inline uint64_t read(serializer, Input& input, rpc::type<int64_t>) { return read_arithmetic_type<int64_t>(input); }
+template <typename Input>
+inline double read(serializer, Input& input, rpc::type<double>) { return read_arithmetic_type<double>(input); }
+
+template <typename Output>
+inline void write(serializer, Output& out, const sstring& v) {
+    write_arithmetic_type(out, uint32_t(v.size()));
+    out.write(v.c_str(), v.size());
+}
+
+template <typename Input>
+inline sstring read(serializer, Input& in, rpc::type<sstring>) {
+    auto size = read_arithmetic_type<uint32_t>(in);
+    sstring ret = uninitialized_string(size);
+    in.read(ret.data(), size);
+    return ret;
+}
+
+namespace bpo = boost::program_options;
+using namespace std::chrono_literals;
+
+class mycomp : public rpc::compressor::factory {
+    const sstring _name = "LZ4";
+public:
+    virtual const sstring& supported() const override {
+        fmt::print("supported called\n");
+        return _name;
+    }
+    virtual std::unique_ptr<rpc::compressor> negotiate(sstring feature, bool is_server) const override {
+        fmt::print("negotiate called with {}\n", feature);
+        return feature == _name ? std::make_unique<rpc::lz4_compressor>() : nullptr;
+    }
+};
+
+int main(int ac, char** av) {
+    app_template app;
+    app.add_options()
+                    ("port", bpo::value<uint16_t>()->default_value(10000), "RPC server port")
+                    ("server", bpo::value<std::string>(), "Server address")
+                    ("compress", bpo::value<bool>()->default_value(false), "Compress RPC traffic");
+    std::cout << "start ";
+    rpc::protocol<serializer> myrpc(serializer{});
+    static std::unique_ptr<rpc::protocol<serializer>::server> server;
+    static std::unique_ptr<rpc::protocol<serializer>::client> client;
+    static double x = 30.0;
+
+    static logger log("rpc_demo");
+    myrpc.set_logger(&log);
+
+    return app.run_deprecated(ac, av, [&] {
+        auto&& config = app.configuration();
+        uint16_t port = config["port"].as<uint16_t>();
+        bool compress = config["compress"].as<bool>();
+        static mycomp mc;
+        auto test1 = myrpc.register_handler(1, [x = 0](int i) mutable { fmt::print("test1 count {:d} got {:d}\n", ++x, i); });
+        auto test2 = myrpc.register_handler(2, [](int a, int b){ fmt::print("test2 got {:d} {:d}\n", a, b); return make_ready_future<int>(a+b); });
+        auto test3 = myrpc.register_handler(3, [](double x){ fmt::print("test3 got {:f}\n", x); return std::make_unique<double>(sin(x)); });
+        auto test4 = myrpc.register_handler(4, [](){ fmt::print("test4 throw!\n"); throw std::runtime_error("exception!"); });
+        auto test5 = myrpc.register_handler(5, [](){ fmt::print("test5 no wait\n"); return rpc::no_wait; });
+        auto test6 = myrpc.register_handler(6, [](const rpc::client_info& info, int x){ fmt::print("test6 client {}, {:d}\n", inet_ntoa(info.addr.as_posix_sockaddr_in().sin_addr), x); });
+        auto test8 = myrpc.register_handler(8, [](){ fmt::print("test8 sleep for 2 sec\n"); return sleep(2s); });
+        auto test13 = myrpc.register_handler(13, [](){ fmt::print("test13 sleep for 1 msec\n"); return sleep(1ms); });
+        auto test_message_to_big = myrpc.register_handler(14, [](sstring payload){ fmt::print("test message to bit, should not get here"); });
+
+        if (config.count("server")) {
+            std::cout << "client" << std::endl;
+            auto test7 = myrpc.make_client<long (long a, long b)>(7);
+            auto test9 = myrpc.make_client<long (long a, long b)>(9); // do not send optional
+            auto test9_1 = myrpc.make_client<long (long a, long b, int c)>(9); // send optional
+            auto test9_2 = myrpc.make_client<long (long a, long b, int c, long d)>(9); // send more data than handler expects
+            auto test10 = myrpc.make_client<long ()>(10); // receive less then replied
+            auto test10_1 = myrpc.make_client<future<rpc::tuple<long, int>> ()>(10); // receive all
+            auto test11 = myrpc.make_client<future<rpc::tuple<long, rpc::optional<int>>> ()>(11); // receive more then replied
+            auto test12 = myrpc.make_client<void (int sleep_ms, sstring payload)>(12); // large payload vs. server limits
+            auto test_nohandler = myrpc.make_client<void ()>(100000000); // non existing verb
+            auto test_nohandler_nowait = myrpc.make_client<rpc::no_wait_type ()>(100000000); // non existing verb, no_wait call
+            rpc::client_options co;
+            if (compress) {
+                co.compressor_factory = &mc;
+            }
+
+            client = std::make_unique<rpc::protocol<serializer>::client>(myrpc, co, ipv4_addr{config["server"].as<std::string>()});
+
+            auto f = test8(*client, 1500ms).then_wrapped([](future<> f) {
+                try {
+                    f.get();
+                    printf("test8 should not get here!\n");
+                } catch (rpc::timeout_error&) {
+                    printf("test8 timeout!\n");
+                }
+            });
+            for (auto i = 0; i < 100; i++) {
+                fmt::print("iteration={:d}\n", i);
+                (void)test1(*client, 5).then([] (){ fmt::print("test1 ended\n");});
+                (void)test2(*client, 1, 2).then([] (int r) { fmt::print("test2 got {:d}\n", r); });
+                (void)test3(*client, x).then([](double x) { fmt::print("sin={:f}\n", x); });
+                (void)test4(*client).then_wrapped([](future<> f) {
+                    try {
+                        f.get();
+                        fmt::print("test4 your should not see this!\n");
+                    } catch (std::runtime_error& x){
+                        fmt::print("test4 {}\n", x.what());
+                    }
+                });
+                (void)test5(*client).then([] { fmt::print("test5 no wait ended\n"); });
+                (void)test6(*client, 1).then([] { fmt::print("test6 ended\n"); });
+                (void)test7(*client, 5, 6).then([] (long r) { fmt::print("test7 got {:d}\n", r); });
+                (void)test9(*client, 1, 2).then([] (long r) { fmt::print("test9 got {:d}\n", r); });
+                (void)test9_1(*client, 1, 2, 3).then([] (long r) { fmt::print("test9.1 got {:d}\n", r); });
+                (void)test9_2(*client, 1, 2, 3, 4).then([] (long r) { fmt::print("test9.2 got {:d}\n", r); });
+                (void)test10(*client).then([] (long r) { fmt::print("test10 got {:d}\n", r); });
+                (void)test10_1(*client).then([] (rpc::tuple<long, int> r) { fmt::print("test10_1 got {:d} and {:d}\n", std::get<0>(r), std::get<1>(r)); });
+                (void)test11(*client).then([] (rpc::tuple<long, rpc::optional<int> > r) { fmt::print("test11 got {:d} and {:d}\n", std::get<0>(r), bool(std::get<1>(r))); });
+                (void)test_nohandler(*client).then_wrapped([](future<> f) {
+                    try {
+                        f.get();
+                        fmt::print("test_nohandler your should not see this!\n");
+                    } catch (rpc::unknown_verb_error& x){
+                        fmt::print("test_nohandle no such verb\n");
+                    } catch (...) {
+                        fmt::print("incorrect exception!\n");
+                    }
+                });
+                (void)test_nohandler_nowait(*client);
+                auto c = make_lw_shared<rpc::cancellable>();
+                (void)test13(*client, *c).then_wrapped([](future<> f) {
+                    try {
+                        f.get();
+                        fmt::print("test13 shold not get here\n");
+                    } catch(rpc::canceled_error&) {
+                        fmt::print("test13 canceled\n");
+                    } catch(...) {
+                        fmt::print("test13 wrong exception\n");
+                    }
+                });
+                c->cancel();
+                (void)test13(*client, *c).then_wrapped([](future<> f) {
+                    try {
+                        f.get();
+                        fmt::print("test13 shold not get here\n");
+                    } catch(rpc::canceled_error&) {
+                        fmt::print("test13 canceled\n");
+                    } catch(...) {
+                        fmt::print("test13 wrong exception\n");
+                    }
+                });
+                (void)sleep(500us).then([c] { c->cancel(); });
+                (void)test_message_to_big(*client, uninitialized_string(10'000'001)).then_wrapped([](future<> f) {
+                    try {
+                        f.get();
+                        fmt::print("test message to big shold not get here\n");
+                    } catch(std::runtime_error& err) {
+                        fmt::print("test message to big get error {}\n", err.what());
+                    } catch(...) {
+                        fmt::print("test message to big wrong exception\n");
+                    }
+                });
+            }
+            // delay a little for a time-sensitive test
+            (void)sleep(400ms).then([test12] () mutable {
+                // server is configured for 10MB max, throw 25MB worth of requests at it.
+                auto now = rpc::rpc_clock_type::now();
+                return parallel_for_each(boost::irange(0, 25), [test12, now] (int idx) mutable {
+                    return test12(*client, 100, uninitialized_string(1'000'000)).then([idx, now] {
+                        auto later = rpc::rpc_clock_type::now();
+                        auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(later - now);
+                        fmt::print("idx {:d} completed after {:d} ms\n", idx, delta.count());
+                    });
+                }).then([now] {
+                    auto later = rpc::rpc_clock_type::now();
+                    auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(later - now);
+                    fmt::print("test12 completed after {:d} ms (should be ~300)\n", delta.count());
+                });
+            });
+            (void)f.finally([] {
+                return sleep(1s).then([] {
+                    return client->stop().then([] {
+                        return engine().exit(0);
+                    });
+                });
+            });
+        } else {
+            std::cout << "server on port " << port << std::endl;
+            myrpc.register_handler(7, [](long a, long b) mutable {
+                auto p = make_lw_shared<promise<>>();
+                auto t = make_lw_shared<timer<>>();
+                fmt::print("test7 got {:d} {:d}\n", a, b);
+                auto f = p->get_future().then([a, b, t] {
+                    fmt::print("test7 calc res\n");
+                    return a - b;
+                });
+                t->set_callback([p = std::move(p)] () mutable { p->set_value(); });
+                t->arm(1s);
+                return f;
+            });
+            myrpc.register_handler(9, [] (long a, long b, rpc::optional<int> c) {
+                long r = 2;
+                fmt::print("test9 got {:d} {:d} ", a, b);
+                if (c) {
+                    fmt::print("{:d}", c.value());
+                    r++;
+                }
+                fmt::print("\n");
+                return r;
+            });
+            myrpc.register_handler(10, [] {
+                fmt::print("test 10\n");
+                return make_ready_future<rpc::tuple<long, int>>(rpc::tuple<long, int>(1, 2));
+            });
+            myrpc.register_handler(11, [] {
+                fmt::print("test 11\n");
+                return 1ul;
+            });
+            myrpc.register_handler(12, [] (int sleep_ms, sstring payload) {
+                return sleep(std::chrono::milliseconds(sleep_ms)).then([] {
+                    return make_ready_future<>();
+                });
+            });
+
+            rpc::resource_limits limits;
+            limits.bloat_factor = 1;
+            limits.basic_request_size = 0;
+            limits.max_memory = 10'000'000;
+            rpc::server_options so;
+            if (compress) {
+                so.compressor_factory = &mc;
+            }
+            server = std::make_unique<rpc::protocol<serializer>::server>(myrpc, so, ipv4_addr{port}, limits);
+        }
+    });
+
+}