Make block and tx svcs thread safe, notifier disabled (WIP).

include/bitcoin/server/server_node.hpp

@@ -20,19 +20,17 @@
 #ifndef LIBBITCOIN_SERVER_SERVER_NODE_HPP
 #define LIBBITCOIN_SERVER_SERVER_NODE_HPP
 
-#include <cstddef>
-#include <cstdint>
 #include <future>
 #include <memory>
 #include <bitcoin/node.hpp>
 #include <bitcoin/protocol.hpp>
 #include <bitcoin/server/configuration.hpp>
 #include <bitcoin/server/define.hpp>
-////#include <bitcoin/server/services/block_service.hpp>
+#include <bitcoin/server/services/block_service.hpp>
 #include <bitcoin/server/services/heart_service.hpp>
 #include <bitcoin/server/services/query_service.hpp>
-////#include <bitcoin/server/services/trans_service.hpp>
-#include <bitcoin/server/utility/address_notifier.hpp>
+#include <bitcoin/server/services/trans_service.hpp>
+////#include <bitcoin/server/services/address_worker.hpp>
 #include <bitcoin/server/utility/curve_authenticator.hpp>
 
 namespace libbitcoin {
@@ -43,10 +41,6 @@ class BCS_API server_node
 {
 public:
     typedef std::shared_ptr<server_node> ptr;
-    typedef std::function<void(uint32_t, const chain::block::ptr)>
-        block_notify_callback;
-    typedef std::function<void (const chain::transaction&)>
-        transaction_notify_callback;
 
     /// Construct a server node.
     server_node(const configuration& configuration);
@@ -76,50 +70,28 @@ class BCS_API server_node
     /// Server configuration settings.
     virtual const settings& server_settings() const;
 
-    // Subscriptions.
-    // ----------------------------------------------------------------------------
-
-    /// Subscribe to block announcements and reorgs.
-    virtual void subscribe_blocks(block_notify_callback notify_block);
-
-    /// Subscribe to new memory pool transactions.
-    virtual void subscribe_transactions(transaction_notify_callback notify_tx);
-
 private:
-    typedef std::vector<block_notify_callback> block_notify_list;
-    typedef std::vector<transaction_notify_callback> transaction_notify_list;
-
-    bool handle_new_transaction(const code& ec,
-        const chain::point::indexes& unconfirmed,
-        const chain::transaction& tx);
-    bool handle_new_blocks(const code& ec, uint64_t fork_point,
-        const chain::block::ptr_list& new_blocks,
-        const chain::block::ptr_list& replaced_blocks);
-
     void handle_running(const code& ec, result_handler handler);
     void handle_closing(const code& ec, std::promise<code>& wait);
 
-    bool start_heart_services();
     bool start_query_services();
+    bool start_heart_services();
+    bool start_block_services();
+    bool start_trans_services();
     bool start_query_workers(bool secure);
 
     const configuration& configuration_;
-    const size_t last_checkpoint_height_;
 
     // These are thread safe.
     curve_authenticator authenticator_;
     query_service secure_query_service_;
     query_service public_query_service_;
     heart_service secure_heart_service_;
     heart_service public_heart_service_;
-
-    // This is protected by block mutex.
-    block_notify_list block_subscriptions_;
-    mutable upgrade_mutex block_mutex_;
-
-    // This is protected by transaction mutex.
-    transaction_notify_list transaction_subscriptions_;
-    mutable upgrade_mutex transaction_mutex_;
+    block_service secure_block_service_;
+    block_service public_block_service_;
+    trans_service secure_trans_service_;
+    trans_service public_trans_service_;
 };
 
 } // namespace server

include/bitcoin/server/services/block_service.hpp

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2011-2015 libbitcoin developers (see AUTHORS)
+ * Copyright (c) 2011-2016 libbitcoin developers (see AUTHORS)
  *
  * This file is part of libbitcoin-server.
  *
@@ -25,44 +25,56 @@
 #include <bitcoin/protocol.hpp>
 #include <bitcoin/server/define.hpp>
 #include <bitcoin/server/settings.hpp>
-#include <bitcoin/server/utility/curve_authenticator.hpp>
 
 namespace libbitcoin {
 namespace server {
 
 class server_node;
 
+// Subscribe to block acceptances into the long chain.
 class BCS_API block_service
-  : public enable_shared_from_base<block_service>
+  : public bc::protocol::zmq::worker
 {
 public:
     typedef std::shared_ptr<block_service> ptr;
 
-    /// Construct a block endpoint.
+    /// The fixed inprocess worker endpoints.
+    static const config::endpoint public_worker;
+    static const config::endpoint secure_worker;
+
+    /// Construct a block service.
     block_service(bc::protocol::zmq::authenticator& authenticator,
         server_node& node, bool secure);
 
-    /// This class is not copyable.
-    block_service(const block_service&) = delete;
-    void operator=(const block_service&) = delete;
-
-    /// Start the endpoint.
-    bool start();
+    /// Start the service.
+    bool start() override;
 
-    /// Stop the endpoint.
+    /// Stop the service.
     bool stop();
 
+protected:
+    typedef chain::block::ptr_list block_list;
+    typedef bc::protocol::zmq::socket socket;
+
+    virtual bool bind(socket& xpub, socket& xsub);
+    virtual bool unbind(socket& xpub, socket& xsub);
+
+    // Implement the service.
+    virtual void work();
+
 private:
-    void send(uint32_t height, const chain::block::ptr block);
+    bool handle_reorganization(const code& ec, uint64_t fork_point,
+        const block_list& new_blocks, const block_list&);
+    void publish_blocks(uint32_t fork_point, const block_list& blocks);
+    void publish_block(socket& publisher, uint32_t height,
+        const chain::block::ptr block);
 
-    // BUGBUG: This is NOT thread safe.
-    bc::protocol::zmq::socket socket_;
+    const bool secure_;
+    const server::settings& settings_;
 
-    // This is thread safe.
+    // These are thread safe.
+    bc::protocol::zmq::authenticator& authenticator_;
     server_node& node_;
-    const bc::config::endpoint endpoint_;
-    const bool enabled_;
-    const bool secure_;
 };
 
 } // namespace server

include/bitcoin/server/services/heart_service.hpp

@@ -42,14 +42,16 @@ class BCS_API heart_service
         server_node& node, bool secure);
 
 protected:
-    virtual bool bind(bc::protocol::zmq::socket& publisher);
-    virtual bool unbind(bc::protocol::zmq::socket& publisher);
+    typedef bc::protocol::zmq::socket socket;
+
+    virtual bool bind(socket& publisher);
+    virtual bool unbind(socket& publisher);
 
     // Implement the service.
     virtual void work();
 
     // Publish the heartbeat (integrated worker).
-    void publish(uint32_t count, bc::protocol::zmq::socket& socket);
+    void publish(uint32_t count, socket& socket);
 
 private:
     const server::settings& settings_;

include/bitcoin/server/services/query_service.hpp

@@ -45,10 +45,10 @@ class BCS_API query_service
         server_node& node, bool secure);
 
 protected:
-    virtual bool bind(bc::protocol::zmq::socket& router,
-        bc::protocol::zmq::socket& dealer);
-    virtual bool unbind(bc::protocol::zmq::socket& router,
-        bc::protocol::zmq::socket& dealer);
+    typedef bc::protocol::zmq::socket socket;
+
+    virtual bool bind(socket& router, socket& dealer);
+    virtual bool unbind(socket& router, socket& dealer);
 
     // Implement the service.
     virtual void work();

include/bitcoin/server/services/trans_service.hpp

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2011-2015 libbitcoin developers (see AUTHORS)
+ * Copyright (c) 2011-2016 libbitcoin developers (see AUTHORS)
  *
  * This file is part of libbitcoin-server.
  *
@@ -20,49 +20,58 @@
 #ifndef LIBBITCOIN_SERVER_TRANS_SERVICE_HPP
 #define LIBBITCOIN_SERVER_TRANS_SERVICE_HPP
 
-#include <cstdint>
 #include <memory>
 #include <bitcoin/protocol.hpp>
 #include <bitcoin/server/define.hpp>
 #include <bitcoin/server/settings.hpp>
-#include <bitcoin/server/utility/curve_authenticator.hpp>
 
 namespace libbitcoin {
 namespace server {
 
 class server_node;
 
+// Subscribe to transaction acceptances into the transaction memory pool.
 class BCS_API trans_service
-  : public enable_shared_from_base<trans_service>
+  : public bc::protocol::zmq::worker
 {
 public:
     typedef std::shared_ptr<trans_service> ptr;
 
-    /// Construct a transaction endpoint.
+    /// The fixed inprocess worker endpoints.
+    static const config::endpoint public_worker;
+    static const config::endpoint secure_worker;
+
+    /// Construct a transaction service.
     trans_service(bc::protocol::zmq::authenticator& authenticator,
         server_node& node, bool secure);
 
-    /// This class is not copyable.
-    trans_service(const trans_service&) = delete;
-    void operator=(const trans_service&) = delete;
-
-    /// Start the endpoint.
-    bool start();
+    /// Start the service.
+    bool start() override;
 
-    /// Stop the endpoint.
+    /// Stop the service.
     bool stop();
 
+protected:
+    typedef bc::protocol::zmq::socket socket;
+    typedef bc::chain::point::indexes index_list;
+
+    virtual bool bind(socket& xpub, socket& xsub);
+    virtual bool unbind(socket& xpub, socket& xsub);
+
+    // Implement the service.
+    virtual void work();
+
 private:
-    void send(const chain::transaction& tx);
+    bool handle_transaction(const code& ec, const index_list&,
+        const chain::transaction& tx);
+    void publish_transaction(const chain::transaction& tx);
 
-    // BUGBUG: This is NOT thread safe.
-    bc::protocol::zmq::socket socket_;
+    const bool secure_;
+    const server::settings& settings_;
 
-    // This is thread safe.
+    // These are thread safe.
+    bc::protocol::zmq::authenticator& authenticator_;
     server_node& node_;
-    const bc::config::endpoint endpoint_;
-    const bool enabled_;
-    const bool secure_;
 };
 
 } // namespace server

include/bitcoin/server/utility/address_notifier.hpp

@@ -75,6 +75,10 @@ class BCS_API address_notifier
         subscription_locator locator;
     };
 
+
+    ////typedef resubscriber<const chain::transaction&> block_subscriber;
+    ////typedef resubscriber<const chain::transaction&> tx_subscriber;
+
     typedef std::vector<subscription_record> subscription_records;
     typedef std::vector<subscription_locator> subscription_locators;
 

include/bitcoin/server/workers/query_worker.hpp

@@ -47,15 +47,17 @@ class BCS_API query_worker
         server_node& node, bool secure);
 
 protected:
+    typedef bc::protocol::zmq::socket socket;
+
     typedef std::function<void(const incoming&, send_handler)> command_handler;
     typedef std::unordered_map<std::string, command_handler> command_map;
 
-    virtual bool connect(bc::protocol::zmq::socket& router);
-    virtual bool disconnect(bc::protocol::zmq::socket& router);
-
     virtual void attach_interface();
     virtual void attach(const std::string& command, command_handler handler);
-    virtual void query(bc::protocol::zmq::socket& router);
+
+    virtual bool connect(socket& router);
+    virtual bool disconnect(socket& router);
+    virtual void query(socket& router);
 
     // Implement the worker.
     virtual void work();

src/messages/incoming.cpp

@@ -50,7 +50,7 @@ std::string incoming::address()
 code incoming::receive(zmq::socket& socket)
 {
     zmq::message message;
-    auto ec = message.receive(socket);
+    auto ec = socket.receive(message);
 
     if (ec)
         return ec;

src/messages/outgoing.cpp

@@ -88,7 +88,7 @@ std::string outgoing::address()
 
 code outgoing::send(zmq::socket& socket)
 {
-    return message_.send(socket);
+    return socket.send(message_);
 }
 
 } // namespace server