7005465978

Committed 27 Nov 2023 01:09PM UTC coverage: 69.961% (+0.3%) from 69.614%

Build # 7005465978

Build Type

push

github

Committed by

bmerry

Commit Message

Make async_send_heap and async_send_heaps work with completion tokens

When using a completion token rather than a callback function, the
boolean return is no longer available. Unfortunately this requires
having two overloads.

Using this with boost::asio::use_future will greatly simplify the common
case of sending a heap then at a later point blocking until it is sent
or throwing if there was a problem.

Run Details

85 of 86 new or added lines in 2 files covered. (98.84%)

4 existing lines in 2 files now uncovered.

4835 of 6911 relevant lines covered (69.96%)

92414.91 hits per line

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

98.72

/src/unittest_send_completion.cpp

/* Copyright 2023 National Research Foundation (SARAO)
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Lesser 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 Lesser General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * @file
 *
 * Unit tests for send completion tokens.
 */

#include <exception>
#include <memory>
#include <future>
#include <utility>
#include <cstddef>
#include <boost/asio.hpp>
#include <boost/test/unit_test.hpp>
#include <spead2/common_defines.h>
#include <spead2/common_thread_pool.h>
#include <spead2/common_inproc.h>
#include <spead2/send_inproc.h>

namespace spead2::unittest
{

BOOST_AUTO_TEST_SUITE(send)
BOOST_AUTO_TEST_SUITE(completion)

// empty heap: header, 4 standard items, padding item, padding
static constexpr std::size_t heap_size = 9 + 5 * sizeof(item_pointer_t);

class promise_handler
{
    std::promise<item_pointer_t> &promise;

public:
    explicit promise_handler(std::promise<item_pointer_t> &promise) : promise(promise) {}

    void operator()(const boost::system::error_code &ec, item_pointer_t bytes_transferred) const
    {
        if (ec)
            promise.set_exception(std::make_exception_ptr(boost::system::system_error(ec)));
        else
            promise.set_value(bytes_transferred);
    }
};

static bool is_would_block(const boost::system::system_error &ex)
{
    return ex.code() == boost::asio::error::would_block;
}

// Test async_send_heap with a completion handler
BOOST_AUTO_TEST_CASE(async_send_heap_handler)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue});
    spead2::send::heap heap;

    std::promise<item_pointer_t> promise;
    auto future = promise.get_future();
    bool result = stream.async_send_heap(heap, promise_handler(promise));
    BOOST_CHECK_EQUAL(result, true);
    BOOST_CHECK_EQUAL(future.get(), heap_size);
}

// Test async_send_heap with a generic token
BOOST_AUTO_TEST_CASE(async_send_heap_token)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue});
    spead2::send::heap heap;

    std::future<item_pointer_t> future = stream.async_send_heap(heap, boost::asio::use_future);
    BOOST_CHECK_EQUAL(future.get(), heap_size);
}

// Test async_send_heaps with a completion handler
BOOST_AUTO_TEST_CASE(async_send_heaps_handler)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue});
    std::array<spead2::send::heap, 2> heaps;

    std::promise<item_pointer_t> promise;
    auto future = promise.get_future();
    bool result = stream.async_send_heaps(
        heaps.begin(), heaps.end(),
        promise_handler(promise),
        spead2::send::group_mode::SERIAL
    );
    BOOST_CHECK_EQUAL(result, true);
    BOOST_CHECK_EQUAL(future.get(), heap_size * heaps.size());
}

// Test async_send_heaps with a completion token
BOOST_AUTO_TEST_CASE(async_send_heaps_token)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue});
    std::array<spead2::send::heap, 2> heaps;

    std::future<item_pointer_t> future = stream.async_send_heaps(
        heaps.begin(), heaps.end(),
        boost::asio::use_future,
        spead2::send::group_mode::SERIAL
    );
    BOOST_CHECK_EQUAL(future.get(), heap_size * heaps.size());
}

// Test async_send_heaps failure case with a completion handler
BOOST_AUTO_TEST_CASE(async_send_heaps_failure_handler)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue}, spead2::send::stream_config().set_max_heaps(1));
    std::array<spead2::send::heap, 2> heaps;

    std::promise<item_pointer_t> promise;
    auto future = promise.get_future();
    bool result = stream.async_send_heaps(
        heaps.begin(), heaps.end(),
        promise_handler(promise),
        spead2::send::group_mode::SERIAL
    );
    BOOST_CHECK_EQUAL(result, false);
    BOOST_CHECK_EXCEPTION(future.get(), boost::system::system_error, is_would_block);
}

// Test async_send_heaps failure case with a completion token
BOOST_AUTO_TEST_CASE(async_send_heaps_failure_token)
{
    thread_pool tp;
    auto queue = std::make_shared<inproc_queue>();
    spead2::send::inproc_stream stream(tp, {queue}, spead2::send::stream_config().set_max_heaps(1));
    std::array<spead2::send::heap, 2> heaps;

    std::promise<item_pointer_t> promise;
    auto future = promise.get_future();
    bool result = stream.async_send_heaps(
        heaps.begin(), heaps.end(),
        [&promise](const boost::system::error_code &ec, item_pointer_t bytes_transferred)
        {
            if (ec)
                promise.set_exception(std::make_exception_ptr(boost::system::system_error(ec)));
            else
                promise.set_value(bytes_transferred);
        },
        spead2::send::group_mode::SERIAL
    );
    BOOST_CHECK_EQUAL(result, false);
    BOOST_CHECK_EXCEPTION(future.get(), boost::system::system_error, is_would_block);
}

BOOST_AUTO_TEST_SUITE_END()  // completion
BOOST_AUTO_TEST_SUITE_END()  // send

} // namespace spead2::unittest

1	/* Copyright 2023 National Research Foundation (SARAO)
2	*
3	* This program is free software: you can redistribute it and/or modify it under
4	* the terms of the GNU Lesser General Public License as published by the Free
5	* Software Foundation, either version 3 of the License, or (at your option) any
6	* later version.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
10	* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
11	* details.
12	*
13	* You should have received a copy of the GNU Lesser General Public License
14	* along with this program. If not, see <http://www.gnu.org/licenses/>.
15	*/
16
17	/**
18	* @file
19	*
20	* Unit tests for send completion tokens.
21	*/
22
23	#include <exception>
24	#include <memory>
25	#include <future>
26	#include <utility>
27	#include <cstddef>
28	#include <boost/asio.hpp>
29	#include <boost/test/unit_test.hpp>
30	#include <spead2/common_defines.h>
31	#include <spead2/common_thread_pool.h>
32	#include <spead2/common_inproc.h>
33	#include <spead2/send_inproc.h>
34
35	namespace spead2::unittest
36	{
37
38	BOOST_AUTO_TEST_SUITE(send)
39	BOOST_AUTO_TEST_SUITE(completion)
40
41	// empty heap: header, 4 standard items, padding item, padding
42	static constexpr std::size_t heap_size = 9 + 5 * sizeof(item_pointer_t);
43
44	class promise_handler
45	{
46	std::promise<item_pointer_t> &promise;
47
48	public:
49	explicit promise_handler(std::promise<item_pointer_t> &promise) : promise(promise) {}	3✔
50
51	void operator()(const boost::system::error_code &ec, item_pointer_t bytes_transferred) const	3✔
52	{
53	if (ec)	3✔
54	promise.set_exception(std::make_exception_ptr(boost::system::system_error(ec)));	1✔
55	else
56	promise.set_value(bytes_transferred);	2✔
57	}	3✔
58	};
59
60	static bool is_would_block(const boost::system::system_error &ex)	2✔
61	{
62	return ex.code() == boost::asio::error::would_block;	2✔
63	}
64
65	// Test async_send_heap with a completion handler
66	BOOST_AUTO_TEST_CASE(async_send_heap_handler)	2✔
67	{
68	thread_pool tp;	1✔
69	auto queue = std::make_shared<inproc_queue>();	1✔
70	spead2::send::inproc_stream stream(tp, {queue});	3✔
71	spead2::send::heap heap;	1✔
72
73	std::promise<item_pointer_t> promise;	1✔
74	auto future = promise.get_future();	1✔
75	bool result = stream.async_send_heap(heap, promise_handler(promise));	1✔
76	BOOST_CHECK_EQUAL(result, true);	1✔
77	BOOST_CHECK_EQUAL(future.get(), heap_size);	1✔
78	}	1✔
79
80	// Test async_send_heap with a generic token
81	BOOST_AUTO_TEST_CASE(async_send_heap_token)	2✔
82	{
83	thread_pool tp;	1✔
84	auto queue = std::make_shared<inproc_queue>();	1✔
85	spead2::send::inproc_stream stream(tp, {queue});	3✔
86	spead2::send::heap heap;	1✔
87
88	std::future<item_pointer_t> future = stream.async_send_heap(heap, boost::asio::use_future);	1✔
89	BOOST_CHECK_EQUAL(future.get(), heap_size);	1✔
90	}	1✔
91
92	// Test async_send_heaps with a completion handler
93	BOOST_AUTO_TEST_CASE(async_send_heaps_handler)	2✔
94	{
95	thread_pool tp;	1✔
96	auto queue = std::make_shared<inproc_queue>();	1✔
97	spead2::send::inproc_stream stream(tp, {queue});	3✔
98	std::array<spead2::send::heap, 2> heaps;	1✔
99
100	std::promise<item_pointer_t> promise;	1✔
101	auto future = promise.get_future();	1✔
102	bool result = stream.async_send_heaps(	1✔
103	heaps.begin(), heaps.end(),
104	promise_handler(promise),	2✔
105	spead2::send::group_mode::SERIAL
106	);	1✔
107	BOOST_CHECK_EQUAL(result, true);	1✔
108	BOOST_CHECK_EQUAL(future.get(), heap_size * heaps.size());	1✔
109	}	1✔
110
111	// Test async_send_heaps with a completion token
112	BOOST_AUTO_TEST_CASE(async_send_heaps_token)	2✔
113	{
114	thread_pool tp;	1✔
115	auto queue = std::make_shared<inproc_queue>();	1✔
116	spead2::send::inproc_stream stream(tp, {queue});	3✔
117	std::array<spead2::send::heap, 2> heaps;	1✔
118
119	std::future<item_pointer_t> future = stream.async_send_heaps(
120	heaps.begin(), heaps.end(),
121	boost::asio::use_future,
122	spead2::send::group_mode::SERIAL
123	);	1✔
124	BOOST_CHECK_EQUAL(future.get(), heap_size * heaps.size());	1✔
125	}	1✔
126
127	// Test async_send_heaps failure case with a completion handler
128	BOOST_AUTO_TEST_CASE(async_send_heaps_failure_handler)	2✔
129	{
130	thread_pool tp;	1✔
131	auto queue = std::make_shared<inproc_queue>();	1✔
132	spead2::send::inproc_stream stream(tp, {queue}, spead2::send::stream_config().set_max_heaps(1));	3✔
133	std::array<spead2::send::heap, 2> heaps;	1✔
134
135	std::promise<item_pointer_t> promise;	1✔
136	auto future = promise.get_future();	1✔
137	bool result = stream.async_send_heaps(	1✔
138	heaps.begin(), heaps.end(),
139	promise_handler(promise),	2✔
140	spead2::send::group_mode::SERIAL
141	);	1✔
142	BOOST_CHECK_EQUAL(result, false);	1✔
143	BOOST_CHECK_EXCEPTION(future.get(), boost::system::system_error, is_would_block);	2✔
144	}	1✔
145
146	// Test async_send_heaps failure case with a completion token
147	BOOST_AUTO_TEST_CASE(async_send_heaps_failure_token)	2✔
148	{
149	thread_pool tp;	1✔
150	auto queue = std::make_shared<inproc_queue>();	1✔
151	spead2::send::inproc_stream stream(tp, {queue}, spead2::send::stream_config().set_max_heaps(1));	3✔
152	std::array<spead2::send::heap, 2> heaps;	1✔
153
154	std::promise<item_pointer_t> promise;	1✔
155	auto future = promise.get_future();	1✔
156	bool result = stream.async_send_heaps(	2✔
157	heaps.begin(), heaps.end(),
158	[&promise](const boost::system::error_code &ec, item_pointer_t bytes_transferred)	2✔
159	{
160	if (ec)	1✔
161	promise.set_exception(std::make_exception_ptr(boost::system::system_error(ec)));	1✔
162	else
NEW 163	promise.set_value(bytes_transferred);	×
164	},	1✔
165	spead2::send::group_mode::SERIAL
166	);	1✔
167	BOOST_CHECK_EQUAL(result, false);	1✔
168	BOOST_CHECK_EXCEPTION(future.get(), boost::system::system_error, is_would_block);	2✔
169	}	1✔
170
171	BOOST_AUTO_TEST_SUITE_END() // completion
172	BOOST_AUTO_TEST_SUITE_END() // send
173
174	} // namespace spead2::unittest

ska-sa / spead2 / 7005465978

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