Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Enable consistent tx/block publishing #512

Merged
merged 1 commit into from
Jan 9, 2019
Merged

Enable consistent tx/block publishing #512

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions src/services/block_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,19 +74,19 @@ bool block_service::start()
void block_service::work()
{
zmq::socket xpub(authenticator_, role::extended_publisher, external_);
zmq::socket xsub(authenticator_, role::extended_subscriber, internal_);
zmq::socket puller(authenticator_, role::puller, internal_);

// Bind sockets to the service and worker endpoints.
if (!started(bind(xpub, xsub)))
if (!started(bind(xpub, puller)))
return;

// TODO: tap in to failure conditions, such as high water.
// BUGBUG: stop is insufficient to stop the worker, because of relay().
// Relay messages between subscriber and publisher (blocks on context).
relay(xpub, xsub);
relay(xpub, puller);

// Unbind the sockets and exit this thread.
finished(unbind(xpub, xsub));
finished(unbind(xpub, puller));
}

// Bind/Unbind.
Expand Down Expand Up @@ -177,11 +177,11 @@ void block_service::publish_blocks(uint32_t fork_height,
if (stopped())
return;

zmq::socket publisher(authenticator_, role::publisher, internal_);
zmq::socket pusher(authenticator_, role::pusher, internal_);

// Subscriptions are off the pub-sub thread so this must connect back.
// This could be optimized by caching the socket as thread static.
const auto ec = publisher.connect(worker_);
const auto ec = pusher.connect(worker_);

if (ec == error::service_stopped)
return;
Expand All @@ -195,14 +195,14 @@ void block_service::publish_blocks(uint32_t fork_height,
}

for (const auto block: *blocks)
publish_block(publisher, ++fork_height, block);
publish_block(pusher, ++fork_height, block);
}

// [ height:4 ]
// [ block ]
// The payload for block publication is delimited within the zeromq message.
// This is required for compatability and inconsistent with query payloads.
void block_service::publish_block(zmq::socket& publisher, size_t height,
void block_service::publish_block(zmq::socket& pusher, size_t height,
block_const_ptr block)
{
if (stopped())
Expand All @@ -217,7 +217,7 @@ void block_service::publish_block(zmq::socket& publisher, size_t height,
broadcast.enqueue(block->to_data(
system::message::version::level::canonical));

const auto ec = publisher.send(broadcast);
const auto ec = pusher.send(broadcast);

if (ec == error::service_stopped)
return;
Expand Down
14 changes: 7 additions & 7 deletions src/services/transaction_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,19 +73,19 @@ bool transaction_service::start()
void transaction_service::work()
{
zmq::socket xpub(authenticator_, role::extended_publisher, external_);
zmq::socket xsub(authenticator_, role::extended_subscriber, internal_);
zmq::socket puller(authenticator_, role::puller, internal_);

// Bind sockets to the service and worker endpoints.
if (!started(bind(xpub, xsub)))
if (!started(bind(xpub, puller)))
return;

// TODO: tap in to failure conditions, such as high water.
// BUGBUG: stop is insufficient to stop the worker, because of relay().
// Relay messages between subscriber and publisher (blocks on context).
relay(xpub, xsub);
relay(xpub, puller);

// Unbind the sockets and exit this thread.
finished(unbind(xpub, xsub));
finished(unbind(xpub, puller));
}

// Bind/Unbind.
Expand Down Expand Up @@ -175,11 +175,11 @@ void transaction_service::publish_transaction(transaction_const_ptr tx)
if (stopped())
return;

zmq::socket publisher(authenticator_, role::publisher, internal_);
zmq::socket pusher(authenticator_, role::pusher, internal_);

// Subscriptions are off the pub-sub thread so this must connect back.
// This could be optimized by caching the socket as thread static.
auto ec = publisher.connect(worker_);
auto ec = pusher.connect(worker_);

if (ec == error::service_stopped)
return;
Expand All @@ -201,7 +201,7 @@ void transaction_service::publish_transaction(transaction_const_ptr tx)
broadcast.enqueue_little_endian(++sequence_);
broadcast.enqueue(tx->to_data(system::message::version::level::canonical));

ec = publisher.send(broadcast);
ec = pusher.send(broadcast);

if (ec == error::service_stopped)
return;
Expand Down