#882

Committed 31 Aug 2023 07:44PM UTC coverage: 41.798% (-44.7%) from 86.546%

Build # #882

Build Type

push

Committed by

Commit Message

Run Details

19442 of 46514 relevant lines covered (41.8%)

126375.38 hits per line

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

64.29

/components/iostreams/src/server/output_stream.cpp

////////////////////////////////////////////////////////////////////////////////
//  Copyright (c) 2011 Bryce Lelbach
//
//  SPDX-License-Identifier: BSL-1.0
//  Distributed under the Boost Software License, Version 1.0. (See accompanying
//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
////////////////////////////////////////////////////////////////////////////////

#include <hpx/config.hpp>
#include <hpx/modules/functional.hpp>

#include <hpx/modules/io_service.hpp>
#include <hpx/modules/serialization.hpp>
#include <hpx/modules/threading_base.hpp>
#include <hpx/runtime_distributed/runtime_fwd.hpp>

#include <hpx/components/iostreams/server/buffer.hpp>
#include <hpx/components/iostreams/server/output_stream.hpp>

#include <cstdint>
#include <functional>
#include <memory>
#include <utility>

namespace hpx::iostreams::detail {

    void buffer::save(serialization::output_archive& ar, unsigned) const
    {
        bool const valid = (data_.get() && !data_->empty());
        ar << valid;
        if (valid)
        {
            ar & data_;
        }
    }

    void buffer::load(serialization::input_archive& ar, unsigned)
    {
        bool valid = false;
        ar >> valid;
        if (valid)
        {
            ar & data_;
        }
    }
}    // namespace hpx::iostreams::detail

namespace hpx::iostreams::server {
    ///////////////////////////////////////////////////////////////////////////
    void output_stream::call_write_async(std::uint32_t locality_id,
        std::uint64_t count, detail::buffer const& in, hpx::id_type /*this_id*/)
    {
        // Perform the IO operation.
        pending_output_.output(locality_id, count, in, write_f, mtx_);
    }

    void output_stream::write_async(std::uint32_t locality_id,
        std::uint64_t count, detail::buffer const& buf_in)
    {
        // Perform the IO in another OS thread.
        detail::buffer in(buf_in);
        // we need to capture the GID of the component to keep it alive long
        // enough.
        hpx::id_type this_id = this->get_id();
#if ASIO_VERSION >= 103400
        ::asio::post(hpx::get_thread_pool("io_pool")->get_io_service(),
            hpx::bind_front(&output_stream::call_write_async, this, locality_id,
                count, HPX_MOVE(in), HPX_MOVE(this_id)));
#else
        hpx::get_thread_pool("io_pool")->get_io_service().post(
            hpx::bind_front(&output_stream::call_write_async, this, locality_id,
                count, HPX_MOVE(in), HPX_MOVE(this_id)));
#endif
    }

    ///////////////////////////////////////////////////////////////////////////
    void output_stream::call_write_sync(std::uint32_t locality_id,
        std::uint64_t count, detail::buffer const& in,
        threads::thread_id_ref_type caller)
    {
        // Perform the IO operation.
        pending_output_.output(locality_id, count, in, write_f, mtx_);

        // Wake up caller.
        threads::set_thread_state(
            caller.noref(), threads::thread_schedule_state::pending);
    }

    void output_stream::write_sync(std::uint32_t locality_id,
        std::uint64_t count, detail::buffer const& buf_in)
    {
        // Perform the IO in another OS thread.
        detail::buffer in(buf_in);
#if ASIO_VERSION >= 103400
        ::asio::post(hpx::get_thread_pool("io_pool")->get_io_service(),
            hpx::bind_front(&output_stream::call_write_sync, this, locality_id,
                count, std::ref(in),
                threads::thread_id_ref_type(threads::get_outer_self_id())));
#else
        hpx::get_thread_pool("io_pool")->get_io_service().post(
            hpx::bind_front(&output_stream::call_write_sync, this, locality_id,
                count, std::ref(in),
                threads::thread_id_ref_type(threads::get_outer_self_id())));
#endif
        // Sleep until the worker thread wakes us up.
        this_thread::suspend(threads::thread_schedule_state::suspended,
            "output_stream::write_sync");
    }
}    // namespace hpx::iostreams::server

1	////////////////////////////////////////////////////////////////////////////////
2	// Copyright (c) 2011 Bryce Lelbach
3	//
4	// SPDX-License-Identifier: BSL-1.0
5	// Distributed under the Boost Software License, Version 1.0. (See accompanying
6	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7	////////////////////////////////////////////////////////////////////////////////
8
9	#include <hpx/config.hpp>
10	#include <hpx/modules/functional.hpp>
11
12	#include <hpx/modules/io_service.hpp>
13	#include <hpx/modules/serialization.hpp>
14	#include <hpx/modules/threading_base.hpp>
15	#include <hpx/runtime_distributed/runtime_fwd.hpp>
16
17	#include <hpx/components/iostreams/server/buffer.hpp>
18	#include <hpx/components/iostreams/server/output_stream.hpp>
19
20	#include <cstdint>
21	#include <functional>
22	#include <memory>
23	#include <utility>
24
25	namespace hpx::iostreams::detail {
26
27	void buffer::save(serialization::output_archive& ar, unsigned) const
28	{
29	bool const valid = (data_.get() && !data_->empty());
30	ar << valid;
31	if (valid)	×
32	{
33	ar & data_;	×
34	}
35	}	×
36
37	void buffer::load(serialization::input_archive& ar, unsigned)	×
38	{
39	bool valid = false;	×
40	ar >> valid;
41	if (valid)	×
42	{
43	ar & data_;	×
44	}
45	}	×
46	} // namespace hpx::iostreams::detail
47		×
48	namespace hpx::iostreams::server {
49	///////////////////////////////////////////////////////////////////////////	×
50	void output_stream::call_write_async(std::uint32_t locality_id,
51	std::uint64_t count, detail::buffer const& in, hpx::id_type /this_id/)
52	{
53	// Perform the IO operation.
54	pending_output_.output(locality_id, count, in, write_f, mtx_);	32✔
55	}
56
57	void output_stream::write_async(std::uint32_t locality_id,
58	std::uint64_t count, detail::buffer const& buf_in)	32✔
59	{	32✔
60	// Perform the IO in another OS thread.
61	detail::buffer in(buf_in);	32✔
62	// we need to capture the GID of the component to keep it alive long
63	// enough.
64	hpx::id_type this_id = this->get_id();
65	#if ASIO_VERSION >= 103400
66	::asio::post(hpx::get_thread_pool("io_pool")->get_io_service(),
67	hpx::bind_front(&output_stream::call_write_async, this, locality_id,
68	count, HPX_MOVE(in), HPX_MOVE(this_id)));
69	#else	32✔
70	hpx::get_thread_pool("io_pool")->get_io_service().post(	32✔
71	hpx::bind_front(&output_stream::call_write_async, this, locality_id,
72	count, HPX_MOVE(in), HPX_MOVE(this_id)));	32✔
73	#endif
74	}
75		45✔
76	///////////////////////////////////////////////////////////////////////////
77	void output_stream::call_write_sync(std::uint32_t locality_id,
78	std::uint64_t count, detail::buffer const& in,
79	threads::thread_id_ref_type caller)
80	{	45✔
81	// Perform the IO operation.
82	pending_output_.output(locality_id, count, in, write_f, mtx_);
83		45✔
84	// Wake up caller.	45✔
85	threads::set_thread_state(	45✔
86	caller.noref(), threads::thread_schedule_state::pending);
87	}	45✔
88
89	void output_stream::write_sync(std::uint32_t locality_id,
90	std::uint64_t count, detail::buffer const& buf_in)
91	{
92	// Perform the IO in another OS thread.	45✔
93	detail::buffer in(buf_in);	45✔
94	#if ASIO_VERSION >= 103400
95	::asio::post(hpx::get_thread_pool("io_pool")->get_io_service(),	90✔
96	hpx::bind_front(&output_stream::call_write_sync, this, locality_id,
97	count, std::ref(in),
98	threads::thread_id_ref_type(threads::get_outer_self_id())));	45✔
99	#else
100	hpx::get_thread_pool("io_pool")->get_io_service().post(	45✔
101	hpx::bind_front(&output_stream::call_write_sync, this, locality_id,
102	count, std::ref(in),
103	threads::thread_id_ref_type(threads::get_outer_self_id())));
104	#endif
105	// Sleep until the worker thread wakes us up.
106	this_thread::suspend(threads::thread_schedule_state::suspended,
107	"output_stream::write_sync");
108	}
109	} // namespace hpx::iostreams::server

STEllAR-GROUP / hpx / #882

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