42421988994

Committed 07 Nov 2026 10:44PM UTC coverage: 93.582% (-0.3%) from 93.863%

Build # 42421988994

Build Type

Pull #1011

github

Committed by

web-flow

Commit Message

Merge 62b70bb20 into ddc7eabb3

Pull Request Pull Request #1011: Read rpc channel post bodies as json-rpc regardless of content type.

Coverage Stats

70 of 70 new or added lines in 5 files covered. (100.0%)

58 existing lines in 7 files now uncovered.

14567 of 15566 relevant lines covered (93.58%)

2260.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.78

/src/net/proxy_queue.cpp

/**
 * Copyright (c) 2011-2026 libbitcoin developers
 *
 * This file is part of libbitcoin.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/network/net/proxy.hpp>

#include <bitcoin/network/define.hpp>
#include <bitcoin/network/log/log.hpp>

// stackoverflow.com/questions/7754695/boost-asio-async-write-how-to-not-
// interleaving-async-write-calls

namespace libbitcoin {
namespace network {

// Shared pointers required in handler parameters so closures control lifetime.
BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED)
BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)

using namespace std::placeholders;

// Send cycle (send continues until queue is empty).
// ----------------------------------------------------------------------------
// private

// The memory of a queue entry apart from its payload: the writer closure with
// its bound pointers, the queue slot, control blocks and the message object.
// This bounds queue depth implicitly, as a flood of small messages is charged
// its estimated cost, not its smaller wire size.
constexpr size_t entry_size = 256;

// Charges the entry against the backlog, false if refused.
bool proxy::charge(pending& write_) NOEXCEPT
{
    BC_ASSERT(stranded());

    write_.cost = system::ceilinged_add(write_.cost, entry_size);

    const auto idle = queue_.empty() && deferred_.empty();
    if (!idle && system::ceilinged_add(backlog_, write_.cost) >
        maximum_backlog_)
    {
        // zeromq messages are seuqenced, so drops are acceptable.
        if (socket_->zeromq())
        {
            LOGS("Dropped message on congested channel [" << endpoint() << "]");
            write_.handler(error::message_dropped, zero);
            return false;
        }

        LOGS("Stopping congested channel [" << endpoint() << "]");
        stop(error::channel_backlog);
        return false;
    }

    backlog_ = system::ceilinged_add(backlog_, write_.cost);
    return true;
}

// A bound entry is an lvalue, moved to the queue from its binding.
void proxy::do_write(pending& write_, bool notification) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (stopped())
    {
        // Does not queue new work or invoke handler after stop.
        LOGQ("Payload write abort [" << endpoint() << "]");
        return;
    }

    // A solicited write (response) is paced by the request that provoked it.
    // A notification must be charged as recipient can force accumulation.
    if (notification)
    {
        if (!charge(write_))
            return;
    }
    else
    {
        write_.cost = zero;
    }

    const auto started = !queue_.empty();
    queue_.push_back(std::move(write_));

    // Start the asynchronous loop if it wasn't already started.
    if (!started)
        write();
}

void proxy::write() NOEXCEPT
{
    BC_ASSERT(stranded());
    if (queue_.empty())
        return;

    // Invokes oldest writer on the queue, completion invokes handle_write.
    writing_ = true;
    queue_.front().call();
}

void proxy::handle_write(const code& ec, size_t bytes) NOEXCEPT
{
    BC_ASSERT(stranded());
    if (queue_.empty())
        return;

    // Handler precedes pop so that a handler send does not start a second
    // write loop (a non-empty queue defers the start to the pop below).
    const auto handler = std::move(queue_.front().handler);
    handler(ec, bytes);
    backlog_ = system::floored_subtract(backlog_, queue_.front().cost);
    queue_.pop_front();

    // All handlers must be invoked unless stopped, so continue despite code.
    if (!stopped())
        write();
}

// Throttle (sent bytes are allocated time at the configured rate).
// ----------------------------------------------------------------------------
// private
// Applied to every send, so the deferral delays the next queued write.

// Nanoseconds allocated to the transmission of bytes at the given rate.
inline steady_clock::duration to_allocation(size_t bytes,
    uint32_t rate) NOEXCEPT
{
    using namespace system;
    constexpr auto nanos = 1'000'000'000_u64;

    // Overflow implies an unusable rate/size, saturated at the type maximum.
    const auto span = ceilinged_multiply<uint64_t>(bytes, nanos) / rate;
    return nanoseconds{ limit<nanoseconds::rep>(span) };
}

count_handler proxy::counted(count_handler&& handler) NOEXCEPT
{
    return std::bind(&proxy::handle_counted,
        shared_from_this(), _1, _2, std::move(handler));
}

void proxy::count_received(size_t bytes) NOEXCEPT
{
    received_ = system::ceilinged_add(received_.load(), bytes);
}

void proxy::handle_counted(const code& ec, size_t bytes,
    const count_handler& handler) NOEXCEPT
{
    count_received(bytes);
    handler(ec, bytes);
}

count_handler proxy::metered(count_handler&& handler) NOEXCEPT
{
    // Stamped at issue, so that only transmission time is credited against
    // the allocation. Time spent idle between sends earns nothing.
    return std::bind(&proxy::handle_metered,
        shared_from_this(), _1, _2, steady_clock::now(), std::move(handler));
}

steady_clock::duration proxy::unconsumed(size_t bytes,
    const steady_clock::time_point& start) const NOEXCEPT
{
    BC_ASSERT(stranded());

    // Stop is never deferred, and a null throttle implies no rate limit.
    if (!throttle_ || stopped())
        return {};

    const auto allocated = to_allocation(bytes, rate_limit_);
    const auto consumed = steady_clock::now() - start;
    return consumed < allocated ? allocated - consumed :
        steady_clock::duration{};
}

void proxy::handle_metered(const code& ec, size_t bytes,
    const steady_clock::time_point& start,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    sent_ = system::ceilinged_add(sent_.load(), bytes);
    writing();
    writing_ = false;

    // A send that consumed its full allocation is not deferred.
    const auto delay = unconsumed(bytes, start);
    if (is_zero(delay.count()))
    {
        handler(ec, bytes);
        return;
    }

    // Handler is posted to the strand, and fired by stop (canceled).
    throttle_->start(std::bind(&proxy::handle_charge,
        shared_from_this(), _1, ec, bytes, handler), delay);
}

void proxy::handle_charge(const code&, const code& ec, size_t bytes,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // The timer code is discarded, as the send result is what is reported.
    handler(ec, bytes);
}

BC_POP_WARNING()
BC_POP_WARNING()
BC_POP_WARNING()

} // namespace network
} // namespace libbitcoin

1	/**
2	* Copyright (c) 2011-2026 libbitcoin developers
3	*
4	* This file is part of libbitcoin.
5	*
6	* This program is free software: you can redistribute it and/or modify
7	* it under the terms of the GNU Affero General Public License as published by
8	* the Free Software Foundation, either version 3 of the License, or
9	* (at your option) any later version.
10	*
11	* This program is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	* GNU Affero General Public License for more details.
15	*
16	* You should have received a copy of the GNU Affero General Public License
17	* along with this program. If not, see <http://www.gnu.org/licenses/>.
18	*/
19	#include <bitcoin/network/net/proxy.hpp>
20
21	#include <bitcoin/network/define.hpp>
22	#include <bitcoin/network/log/log.hpp>
23
24	// stackoverflow.com/questions/7754695/boost-asio-async-write-how-to-not-
25	// interleaving-async-write-calls
26
27	namespace libbitcoin {
28	namespace network {
29
30	// Shared pointers required in handler parameters so closures control lifetime.
31	BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
32	BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED)
33	BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
34
35	using namespace std::placeholders;
36
37	// Send cycle (send continues until queue is empty).
38	// ----------------------------------------------------------------------------
39	// private
40
41	// The memory of a queue entry apart from its payload: the writer closure with
42	// its bound pointers, the queue slot, control blocks and the message object.
43	// This bounds queue depth implicitly, as a flood of small messages is charged
44	// its estimated cost, not its smaller wire size.
45	constexpr size_t entry_size = 256;
46
47	// Charges the entry against the backlog, false if refused.
48	bool proxy::charge(pending& write_) NOEXCEPT	48 ✔
49	{
50	BC_ASSERT(stranded());	48 ✔
51
52	write_.cost = system::ceilinged_add(write_.cost, entry_size);	48 ✔
53
54	const auto idle = queue_.empty() && deferred_.empty();	48 ✔
55	if (!idle && system::ceilinged_add(backlog_, write_.cost) >	3 ✔
56	maximum_backlog_)	3 ✔
57	{
58	// zeromq messages are seuqenced, so drops are acceptable.
59	if (socket_->zeromq())	2 ✔
60	{
61	LOGS("Dropped message on congested channel [" << endpoint() << "]");	1 ✔
62	write_.handler(error::message_dropped, zero);	1 ✔
63	return false;	1 ✔
64	}
65
66	LOGS("Stopping congested channel [" << endpoint() << "]");	1 ✔
67	stop(error::channel_backlog);	1 ✔
68	return false;	1 ✔
69	}
70
71	backlog_ = system::ceilinged_add(backlog_, write_.cost);	46 ✔
72	return true;	46 ✔
73	}
74
75	// A bound entry is an lvalue, moved to the queue from its binding.
76	void proxy::do_write(pending& write_, bool notification) NOEXCEPT	421 ✔
77	{
78	BC_ASSERT(stranded());	421 ✔
79
80	if (stopped())	421 ✔
81	{
82	// Does not queue new work or invoke handler after stop.
83	LOGQ("Payload write abort [" << endpoint() << "]");	2 ✔
84	return;	2 ✔
85	}
86
87	// A solicited write (response) is paced by the request that provoked it.
88	// A notification must be charged as recipient can force accumulation.
89	if (notification)	419 ✔
90	{
91	if (!charge(write_))	47 ✔
92	return;
93	}
94	else
95	{
96	write_.cost = zero;	372 ✔
97	}
98
99	const auto started = !queue_.empty();	417 ✔
100	queue_.push_back(std::move(write_));	417 ✔
101
102	// Start the asynchronous loop if it wasn't already started.
103	if (!started)	417 ✔
104	write();	374 ✔
105	}
106
107	void proxy::write() NOEXCEPT	762 ✔
108	{
109	BC_ASSERT(stranded());	762 ✔
110	if (queue_.empty())	762 ✔
111	return;
112
113	// Invokes oldest writer on the queue, completion invokes handle_write.
114	writing_ = true;	417 ✔
115	queue_.front().call();	417 ✔
116	}
117
118	void proxy::handle_write(const code& ec, size_t bytes) NOEXCEPT	417 ✔
119	{
120	BC_ASSERT(stranded());	417 ✔
121	if (queue_.empty())	417 ✔
UNCOV 122	return;	×
123
124	// Handler precedes pop so that a handler send does not start a second
125	// write loop (a non-empty queue defers the start to the pop below).
126	const auto handler = std::move(queue_.front().handler);	417 ✔
127	handler(ec, bytes);	417 ✔
128	backlog_ = system::floored_subtract(backlog_, queue_.front().cost);	417 ✔
129	queue_.pop_front();	417 ✔
130
131	// All handlers must be invoked unless stopped, so continue despite code.
132	if (!stopped())	417 ✔
133	write();	388 ✔
134	}	417 ✔
135
136	// Throttle (sent bytes are allocated time at the configured rate).
137	// ----------------------------------------------------------------------------
138	// private
139	// Applied to every send, so the deferral delays the next queued write.
140
141	// Nanoseconds allocated to the transmission of bytes at the given rate.
142	inline steady_clock::duration to_allocation(size_t bytes,	4 ✔
143	uint32_t rate) NOEXCEPT
144	{
145	using namespace system;
146	constexpr auto nanos = 1'000'000'000_u64;
147
148	// Overflow implies an unusable rate/size, saturated at the type maximum.
149	const auto span = ceilinged_multiply<uint64_t>(bytes, nanos) / rate;	4 ✔
150	return nanoseconds{ limit<nanoseconds::rep>(span) };	4 ✔
151	}
152
153	count_handler proxy::counted(count_handler&& handler) NOEXCEPT	386 ✔
154	{
155	return std::bind(&proxy::handle_counted,	772 ✔
156	shared_from_this(), _1, _2, std::move(handler));	1,158 ✔
157	}
158
159	void proxy::count_received(size_t bytes) NOEXCEPT	464 ✔
160	{
161	received_ = system::ceilinged_add(received_.load(), bytes);	464 ✔
162	}	464 ✔
163
164	void proxy::handle_counted(const code& ec, size_t bytes,	386 ✔
165	const count_handler& handler) NOEXCEPT
166	{
167	count_received(bytes);	386 ✔
168	handler(ec, bytes);	386 ✔
169	}	386 ✔
170
171	count_handler proxy::metered(count_handler&& handler) NOEXCEPT	418 ✔
172	{
173	// Stamped at issue, so that only transmission time is credited against
174	// the allocation. Time spent idle between sends earns nothing.
175	return std::bind(&proxy::handle_metered,	836 ✔
176	shared_from_this(), _1, _2, steady_clock::now(), std::move(handler));	1,254 ✔
177	}
178
179	steady_clock::duration proxy::unconsumed(size_t bytes,	423 ✔
180	const steady_clock::time_point& start) const NOEXCEPT
181	{
182	BC_ASSERT(stranded());	423 ✔
183
184	// Stop is never deferred, and a null throttle implies no rate limit.
185	if (!throttle_ \|\| stopped())	423 ✔
186	return {};	419 ✔
187
188	const auto allocated = to_allocation(bytes, rate_limit_);	4 ✔
189	const auto consumed = steady_clock::now() - start;	4 ✔
190	return consumed < allocated ? allocated - consumed :	4 ✔
191	steady_clock::duration{};	1 ✔
192	}
193
194	void proxy::handle_metered(const code& ec, size_t bytes,	418 ✔
195	const steady_clock::time_point& start,
196	const count_handler& handler) NOEXCEPT
197	{
198	BC_ASSERT(stranded());	418 ✔
199	sent_ = system::ceilinged_add(sent_.load(), bytes);	836 ✔
200	writing();	418 ✔
201	writing_ = false;	418 ✔
202
203	// A send that consumed its full allocation is not deferred.
204	const auto delay = unconsumed(bytes, start);	418 ✔
205	if (is_zero(delay.count()))	418 ✔
206	{
207	handler(ec, bytes);	417 ✔
208	return;	417 ✔
209	}
210
211	// Handler is posted to the strand, and fired by stop (canceled).
212	throttle_->start(std::bind(&proxy::handle_charge,	1 ✔
213	shared_from_this(), _1, ec, bytes, handler), delay);	2 ✔
214	}
215
216	void proxy::handle_charge(const code&, const code& ec, size_t bytes,	1 ✔
217	const count_handler& handler) NOEXCEPT
218	{
219	BC_ASSERT(stranded());	1 ✔
220
221	// The timer code is discarded, as the send result is what is reported.
222	handler(ec, bytes);	1 ✔
223	}	1 ✔
224
225	BC_POP_WARNING()
226	BC_POP_WARNING()
227	BC_POP_WARNING()
228
229	} // namespace network
230	} // namespace libbitcoin

libbitcoin / libbitcoin-network / 42421988994

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous