Merge pull request #24 from basiliscos/issue-22

Let drop_result policy be useable

Author: basiliscos

CMakeLists.txt

@@ -7,6 +7,8 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 include (CTest)
 enable_testing()
 
+
+
 add_definitions(-DBOOST_ERROR_CODE_HEADER_ONLY)
 add_definitions(-DBOOST_COROUTINES_NO_DEPRECATION_WARNING)
 #add_definitions(-DBREDIS_DEBUG)
@@ -70,7 +72,9 @@ if(WIN32)
     set(LINK_DEPENDENCIES ${Boost_LIBRARIES} catch_lib)
     add_definitions(-DBOOST_ALL_DYN_LINK -DBOOST_THREAD_USE_LIB -D_WIN32_WINNT=0x0501)
 else()
-    set(LINK_DEPENDENCIES pthread ${Boost_LIBRARIES} catch_lib)
+    add_definitions(-Wall -Wextra -pedantic -Werror)
+    #add_definitions(-fsanitize=undefined -fsanitize=address -fno-omit-frame-pointer)
+    set(LINK_DEPENDENCIES pthread ${Boost_LIBRARIES} catch_lib) #  asan ubsan
 endif()
 
 add_subdirectory(examples)
@@ -141,6 +145,10 @@ add_executable(t-21-coroutine t/21-coroutine.cpp)
 target_link_libraries(t-21-coroutine ${LINK_DEPENDENCIES})
 add_test("t-21-coroutine" t-21-coroutine)
 
+add_executable(t-22-ping_drop-policy t/22-ping_drop-policy.cpp)
+target_link_libraries(t-22-ping_drop-policy ${LINK_DEPENDENCIES})
+add_test("t-22-ping_drop-policy" t-22-ping_drop-policy)
+
 add_executable(t-23-stream t/23-stream.cpp)
 target_link_libraries(t-23-stream ${LINK_DEPENDENCIES})
 add_test("t-23-stream" t-23-stream)

README.md

@@ -57,15 +57,19 @@ can be done on the socket object outside of the connector)
 
 Results achieved with `examples/speed_test_async_multi.cpp` for 1 thread, Intel Core i7-4800MQ, gentoo linux
 
-|  bredis (commands/s)       | redox (commands/s)           |
-|----------------------------|------------------------------|
-|  1.30257e+06               |  1.19214e+06                 |
+| bredis (commands/s) | bredis(*) (commands/s) | redox (commands/s)
+|---------------------+------------------------+--------------------
+|      1.59325e+06    |      2.50826e+06       |    0.999375+06
 
 These results are not completely fair, because of the usage of different semantics in the
 APIs; however they are still interesting, as they are using different
 underlying event libraries ([Boost::ASIO](http://www.boost.org/doc/libs/release/libs/asio/) vs [libev](http://software.schmorp.de/pkg/libev.html)) as well as redis protocol
 parsing libraries (written from scratch vs [hiredis](https://github.com/redis/hiredis))
 
+`(*)` bredis with drop_result policy, i.e. replies from redis server are
+scanned only for formal correctness and never delivered to the caller.
+
+
 ## Work with the result
 
 The general idea is that the result of trying to parse a redis reply can be either: not enough data, protocol error (in an extreme case) or some positive parse result. The last one is just **markers** of the result, which is actually stored in the *receive buffer* (i.e. outside of markers, and outside of the bredis-connection).
@@ -386,7 +390,7 @@ boost::asio::spawn(
 ## Steams
 
 There is no specific support for streams (appeared in redis 5.0) in bredis,
-they are just usual `XADD`, `XRANGE` etc. commands and corresponding replies. 
+they are just usual `XADD`, `XRANGE` etc. commands and corresponding replies.
 
 ```cpp
 ...
@@ -410,7 +414,7 @@ rx_buff.consume(parse_result3.consumed);
 auto& outer_arr = boost::get<r::extracts::array_holder_t>(extract3);
 auto& inner_arr1 = boost::get<r::extracts::array_holder_t>(outer_arr.elements[0]);
 auto& inner_arr2 = boost::get<r::extracts::array_holder_t>(outer_arr.elements[1]);
-... 
+...
 
 ```
 
@@ -441,6 +445,35 @@ r::Connection<next_layer_t> c(socket);
 socket.cancel();
 ```
 
+## Thread-safety
+
+`bredis` itself is thread-agnostic, however the underlying socket (`next_layer_t`)
+and used buffers are usually not thread-safe. To handle that in multi-thead
+environment the access to those objects should be sequenced via
+`asio::io_context::strand` . See the `examples/multi-threads-1.cpp`.
+
+
+## parsing_policy::drop_result
+The performance still can be boosted if it is known beforehand that the response from
+redis server is not needed at all. For example, the only possible response to `PING`
+command is `PONG` reply, usually there is no sense it validating that `PONG` reply,
+as soon as it is known, that redis-server alredy delivered us **some** reply
+(in practice it is `PONG`). Another example is `SET` command, when redis-server
+**usually** replies with `OK`.
+
+With `parsing_policy::drop_result` the reply result is just verified with formal
+compliance to redis protocol, and then it is discarded.
+
+It should be noted, that redis can reply back with error, which aslo correct
+reply, but the caller side isn't able to see it when `parsing_policy::drop_result`
+is applied. So, it should be used with care, when you know what your are doing. You have
+been warned.
+
+It is safe, however, to mix different parsing policies on the same connection,
+i.e. write `SET` command and read it's reply with `parsing_policy::drop_result` and
+then write `GET` command and read it's reply with `parsing_policy::keep_result`.
+See the `examples/speed_test_async_multi.cpp`.
+
 ## API
 
 There's a convenience header include/bredis.hpp, doing `#include "bredis.hpp"` will include
@@ -675,7 +708,7 @@ The asynchronous read has the following signature:
 ```cpp
 void-or-deduced
 async_read(DynamicBuffer &rx_buff, ReadCallback read_callback,
-               std::size_t replies_count = 1);
+               std::size_t replies_count = 1, Policy = bredis::parsing_policy::keep_result{});
 ```
 
 It reads `replies_count` replies from the *next_layer* stream, which will be

examples/CMakeLists.txt

@@ -1,13 +1,15 @@
 find_package(Threads REQUIRED)
 
+set(LINK_DEPENDENCIES ${Boost_LIBRARIES} Threads::Threads)
+
 add_executable(stream-parse stream-parse.cpp)
-target_link_libraries(stream-parse ${Boost_LIBRARIES} Threads::Threads)
+target_link_libraries(stream-parse ${LINK_DEPENDENCIES})
 
 add_executable(synch-subscription synch-subscription.cpp)
-target_link_libraries(synch-subscription ${Boost_LIBRARIES} Threads::Threads)
+target_link_libraries(synch-subscription ${LINK_DEPENDENCIES})
 
 add_executable(speed_test_async_multi speed_test_async_multi.cpp)
-target_link_libraries(speed_test_async_multi ${Boost_LIBRARIES} Threads::Threads)
+target_link_libraries(speed_test_async_multi ${LINK_DEPENDENCIES})
 
 add_executable(multi-threads-1 multi-threads-1.cpp)
-target_link_libraries(multi-threads-1 ${Boost_LIBRARIES} Threads::Threads)
+target_link_libraries(multi-threads-1 ${LINK_DEPENDENCIES})

examples/speed_test_async_multi.cpp

@@ -1,22 +1,26 @@
 //
 //
-// Copyright (c) 2017 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
+// Copyright (c) 2017-2019 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
 //
 // Distributed under the MIT Software License
 //
 // mimics performance measurements from
 // https://github.com/hmartiro/redox/blob/master/examples/speed_test_async_multi.cpp
 //
-// Results (1 thread, Intel Core i7-4800MQ, linux)
+// Results (1 thread, Intel Core i7-8550U, void-linux, gcc 8.3.0)
 //
-//   bredis (commands/s)   |  redox (commands/s)
-//  -----------------------+-----------------------
-//        1.30257e+06      |    1.19214e+06
+//  bredis (commands/s) | bredis(*) (commands/s) | redox (commands/s)
+// ---------------------+------------------------+---------------------
+//       1.59325e+06    |      2.50826e+06       |    0.999375+06
 //
 // Results are not completely fair, because of usage of different semantics in
 // APIs; however they are still interesting, as there are used different
 // underlying event libraries (Boost::ASIO vs libev) as well redis protocol
 // parsing library (written from scratch vs hiredis)
+//
+// (*) bredis with drop_result policy, i.e. replies from redis server are
+// scanned only for formal correctness and never delivered to the caller
+
 
 #include <algorithm>
 #include <atomic>
@@ -43,15 +47,18 @@ double time_s() {
 // alias namespaces
 namespace r = bredis;
 namespace asio = boost::asio;
+using boost::get;
 
 int main(int argc, char **argv) {
     // common setup
     using socket_t = asio::ip::tcp::socket;
     using next_layer_t = socket_t;
     using Buffer = boost::asio::streambuf;
     using Iterator = typename r::to_iterator<Buffer>::iterator_t;
+    using policy_t = r::parsing_policy::drop_result;
+    //using policy_t = r::parsing_policy::keep_result;
 
-    if (argc < 1) {
+    if (argc < 2) {
         std::cout << "Usage : " << argv[0] << " ip:port \n";
         return 1;
     }
@@ -70,12 +77,10 @@ int main(int argc, char **argv) {
 
     // write subscribe cmd
     r::single_command_t cmd_incr{"INCR", "simple_loop:count"};
-    r::single_command_t cmd_get{"GET", "simple_loop:count"};
     r::command_container_t cmd_container;
-    for (auto i = 0; i < cmds_count; ++i) {
+    for (size_t i = 0; i < cmds_count; ++i) {
         cmd_container.push_back(cmd_incr);
     }
-    cmd_container.push_back(cmd_get);
 
     r::command_wrapper_t cmd_wpapper{std::move(cmd_container)};
 
@@ -92,29 +97,27 @@ int main(int argc, char **argv) {
     r::Connection<next_layer_t> c(std::move(socket));
 
     Buffer tx_buff, rx_buff;
-    std::promise<std::string> completion_promise;
-    std::future<std::string> completion_future =
-        completion_promise.get_future();
+    std::promise<void> completion_promise;
+    auto completion_future = completion_promise.get_future();
 
     c.async_read(
         rx_buff,
         [&](const boost::system::error_code &ec, auto &&r) {
             assert(!ec);
-            auto &replies =
-                boost::get<r::markers::array_holder_t<Iterator>>(r.result);
-            auto &last_reply = replies.elements.at(replies.elements.size() - 1);
-            auto &str_reply =
-                boost::get<r::markers::string_t<Iterator>>(last_reply);
-            std::string value{str_reply.from, str_reply.to};
+            (void)ec;
             rx_buff.consume(r.consumed);
-            count += replies.elements.size() - 1;
-            completion_promise.set_value(value);
+            // cannot be done with drop_result
+            //auto &replies = get<r::markers::array_holder_t<Iterator>>(r.result);
+            //count += replies.elements.size() - 1;
+            count = cmds_count;
+            completion_promise.set_value();
             std::cout << "done reading...\n";
         },
-        cmds_count + 1);
+    cmds_count, policy_t{});
 
     c.async_write(tx_buff, cmd_wpapper, [&](const boost::system::error_code &ec,
                                             auto bytes_transferred) {
+        (void)ec;
         assert(!ec);
         tx_buff.consume(bytes_transferred);
         std::cout << "done writing...\n";
@@ -130,13 +133,20 @@ int main(int argc, char **argv) {
 
     io_service.run();
     std::cout << "done...\n";
+    completion_future.get();
+
+    c.write(r::single_command_t{"GET", "simple_loop:count"});
+    auto r = c.read(rx_buff);
+    auto &str_reply = get<r::markers::string_t<Iterator>>(r.result);
+
+    std::string counter_value {str_reply.from, str_reply.to};
 
     double actual_freq = (double)count / t_elapsed;
     std::cout << "Sent " << cmds_count << " commands in " << t_elapsed << "s, "
               << "that's " << actual_freq << " commands/s."
               << "\n";
 
-    std::cout << "Final value of counter: " << completion_future.get() << "\n";
+    std::cout << "Final value of counter: " << counter_value << "\n";
 
     std::cout << "exiting...\n";
     return 0;

examples/stream-parse.cpp

@@ -56,7 +56,7 @@ using option_t = boost::optional<json_payload>;
 
 template <typename Iterator>
 struct json_extractor : public boost::static_visitor<option_t> {
-    template <typename T> option_t operator()(const T &value) const {
+    template <typename T> option_t operator()(const T & /*value*/) const {
         return option_t{};
     }
 

examples/synch-subscription.cpp

@@ -57,7 +57,7 @@ using optional_string_t =
     boost::optional<std::pair<string_wrapper_t, string_wrapper_t>>;
 
 struct payload_extractor : public boost::static_visitor<optional_string_t> {
-    template <typename T> optional_string_t operator()(const T &value) const {
+    template <typename T> optional_string_t operator()(const T & /* value */) const {
         return optional_string_t{};
     }
 

include/bredis.hpp

@@ -4,8 +4,7 @@
 //
 // Distributed under the MIT Software License
 //
-#ifndef BREDIS_HPP
-#define BREDIS_HPP
+#pragma once
 
 #include <bredis/Command.hpp>
 #include <bredis/Connection.hpp>
@@ -15,5 +14,3 @@
 #include <bredis/Markers.hpp>
 #include <bredis/Protocol.hpp>
 #include <bredis/Result.hpp>
-
-#endif // BREDIS_HPP

include/bredis/Connection.hpp

@@ -1,6 +1,7 @@
 //
 //
-// Copyright (c) 2017 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
+// Copyright (c) 2017, 2019 Ivan Baidakou (basiliscos) (the dot dmol at gmail
+// dot com)
 //
 // Distributed under the MIT Software License
 //
@@ -27,9 +28,8 @@
 
 namespace bredis {
 
-#define BREDIS_PARSE_RESULT(B)                                                 \
-    positive_parse_result_t<typename to_iterator<B>::iterator_t,               \
-                            bredis::parsing_policy::keep_result>
+#define BREDIS_PARSE_RESULT(B, P)                                              \
+    positive_parse_result_t<typename to_iterator<B>::iterator_t, P>
 
 template <typename NextLayer> class Connection {
 
@@ -51,23 +51,25 @@ template <typename NextLayer> class Connection {
     async_write(DynamicBuffer &tx_buff, const command_wrapper_t &command,
                 WriteCallback &&write_callback);
 
-    template <typename DynamicBuffer, typename ReadCallback>
+    template <typename DynamicBuffer, typename ReadCallback,
+              typename Policy = bredis::parsing_policy::keep_result>
     BOOST_ASIO_INITFN_RESULT_TYPE(ReadCallback,
                                   void(boost::system::error_code,
-                                       BREDIS_PARSE_RESULT(DynamicBuffer)))
+                                       BREDIS_PARSE_RESULT(DynamicBuffer,
+                                                           Policy)))
     async_read(DynamicBuffer &rx_buff, ReadCallback &&read_callback,
-               std::size_t replies_count = 1);
+               std::size_t replies_count = 1, Policy policy = Policy{});
 
     /* synchronous interface */
     void write(const command_wrapper_t &command);
     void write(const command_wrapper_t &command, boost::system::error_code &ec);
 
     template <typename DynamicBuffer>
-    BREDIS_PARSE_RESULT(DynamicBuffer)
+    BREDIS_PARSE_RESULT(DynamicBuffer, bredis::parsing_policy::keep_result)
     read(DynamicBuffer &rx_buff);
 
     template <typename DynamicBuffer>
-    BREDIS_PARSE_RESULT(DynamicBuffer)
+    BREDIS_PARSE_RESULT(DynamicBuffer, bredis::parsing_policy::keep_result)
     read(DynamicBuffer &rx_buff, boost::system::error_code &ec);
 };
 

include/bredis/Extract.hpp

@@ -78,7 +78,7 @@ struct extractor : public boost::static_visitor<extracts::extraction_result_t> {
     }
 
     extracts::extraction_result_t
-    operator()(const markers::nil_t<Iterator> &value) const {
+    operator()(const markers::nil_t<Iterator> & /*value*/) const {
         return extracts::nil_t{};
     }