Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
889ec7ef
Unverified
Commit
889ec7ef
authored
Apr 19, 2020
by
Dominik Charousset
Committed by
GitHub
Apr 19, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #1090
Use only constexpr values as defaults
parents
e74fd805
f67373d2
Changes
9
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
106 additions
and
199 deletions
+106
-199
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+0
-1
libcaf_core/caf/defaults.hpp
libcaf_core/caf/defaults.hpp
+61
-48
libcaf_core/caf/scheduler/abstract_coordinator.hpp
libcaf_core/caf/scheduler/abstract_coordinator.hpp
+2
-0
libcaf_core/src/actor_system_config.cpp
libcaf_core/src/actor_system_config.cpp
+2
-3
libcaf_core/src/defaults.cpp
libcaf_core/src/defaults.cpp
+0
-127
libcaf_core/src/scheduler/abstract_coordinator.cpp
libcaf_core/src/scheduler/abstract_coordinator.cpp
+9
-1
libcaf_core/test/thread_hook.cpp
libcaf_core/test/thread_hook.cpp
+4
-4
libcaf_io/src/io/basp/instance.cpp
libcaf_io/src/io/basp/instance.cpp
+20
-7
libcaf_io/test/io/basp_broker.cpp
libcaf_io/test/io/basp_broker.cpp
+8
-8
No files found.
libcaf_core/CMakeLists.txt
View file @
889ec7ef
...
@@ -72,7 +72,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
...
@@ -72,7 +72,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/credit_controller.cpp
src/credit_controller.cpp
src/decorator/sequencer.cpp
src/decorator/sequencer.cpp
src/default_attachable.cpp
src/default_attachable.cpp
src/defaults.cpp
src/deserializer.cpp
src/deserializer.cpp
src/detail/abstract_worker.cpp
src/detail/abstract_worker.cpp
src/detail/abstract_worker_hub.cpp
src/detail/abstract_worker_hub.cpp
...
...
libcaf_core/caf/defaults.hpp
View file @
889ec7ef
...
@@ -20,77 +20,90 @@
...
@@ -20,77 +20,90 @@
#include <chrono>
#include <chrono>
#include <cstddef>
#include <cstddef>
#include <limits>
#include <string>
#include <string>
#include <vector>
#include <vector>
#include "caf/detail/core_export.hpp"
#include "caf/detail/build_config.hpp"
#include "caf/detail/log_level.hpp"
#include "caf/string_view.hpp"
#include "caf/string_view.hpp"
#include "caf/timestamp.hpp"
#include "caf/timestamp.hpp"
// -- hard-coded default values for various CAF options ------------------------
// -- hard-coded default values for various CAF options ------------------------
namespace
caf
::
defaults
{
namespace
caf
::
defaults
::
stream
{
namespace
stream
{
constexpr
auto
desired_batch_complexity
=
timespan
{
50'000
};
constexpr
auto
max_batch_delay
=
timespan
{
5'000'000
};
constexpr
auto
credit_round_interval
=
timespan
{
10'000'000
};
constexpr
auto
credit_policy
=
string_view
{
"complexity"
};
extern
CAF_CORE_EXPORT
const
timespan
desired_batch_complexity
;
}
// namespace caf::defaults::stream
extern
CAF_CORE_EXPORT
const
timespan
max_batch_delay
;
extern
CAF_CORE_EXPORT
const
timespan
credit_round_interval
;
extern
CAF_CORE_EXPORT
const
string_view
credit_policy
;
namespace
size_policy
{
namespace
caf
::
defaults
::
stream
::
size_policy
{
extern
CAF_CORE_EXPORT
const
int32_t
bytes_per_batch
;
constexpr
auto
bytes_per_batch
=
int32_t
{
02
*
1024
};
// 2 KB
extern
CAF_CORE_EXPORT
const
int32_t
buffer_capacity
;
constexpr
auto
buffer_capacity
=
int32_t
{
64
*
1024
};
// 64 KB
}
// namespace size_policy
}
// namespace
caf::defaults::stream::
size_policy
}
// namespace stream
namespace
caf
::
defaults
::
scheduler
{
namespace
scheduler
{
constexpr
auto
policy
=
string_view
{
"stealing"
};
constexpr
auto
profiling_output_file
=
string_view
{
""
};
constexpr
auto
max_throughput
=
std
::
numeric_limits
<
size_t
>::
max
();
constexpr
auto
profiling_resolution
=
timespan
(
100'000'000
);
extern
CAF_CORE_EXPORT
const
string_view
policy
;
}
// namespace caf::defaults::scheduler
extern
CAF_CORE_EXPORT
string_view
profiling_output_file
;
extern
CAF_CORE_EXPORT
const
size_t
max_threads
;
extern
CAF_CORE_EXPORT
const
size_t
max_throughput
;
extern
CAF_CORE_EXPORT
const
timespan
profiling_resolution
;
}
// namespace scheduler
namespace
caf
::
defaults
::
work_stealing
{
namespace
work_stealing
{
constexpr
auto
aggressive_poll_attempts
=
size_t
{
100
};
constexpr
auto
aggressive_steal_interval
=
size_t
{
10
};
constexpr
auto
moderate_poll_attempts
=
size_t
{
500
};
constexpr
auto
moderate_steal_interval
=
size_t
{
5
};
constexpr
auto
moderate_sleep_duration
=
timespan
{
50'000
};
constexpr
auto
relaxed_steal_interval
=
size_t
{
1
};
constexpr
auto
relaxed_sleep_duration
=
timespan
{
10'000'000
};
extern
CAF_CORE_EXPORT
const
size_t
aggressive_poll_attempts
;
}
// namespace caf::defaults::work_stealing
extern
CAF_CORE_EXPORT
const
size_t
aggressive_steal_interval
;
extern
CAF_CORE_EXPORT
const
size_t
moderate_poll_attempts
;
extern
CAF_CORE_EXPORT
const
size_t
moderate_steal_interval
;
extern
CAF_CORE_EXPORT
const
timespan
moderate_sleep_duration
;
extern
CAF_CORE_EXPORT
const
size_t
relaxed_steal_interval
;
extern
CAF_CORE_EXPORT
const
timespan
relaxed_sleep_duration
;
}
// namespace work_stealing
namespace
caf
::
defaults
::
logger
{
namespace
logger
{
constexpr
auto
default_log_level
=
string_view
{
#if CAF_LOG_LEVEL == CAF_LOG_LEVEL_TRACE
"trace"
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_DEBUG
"debug"
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_INFO
"info"
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_WARNING
"warning"
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_ERROR
"error"
#else
"quiet"
#endif
};
extern
CAF_CORE_EXPORT
string_view
component_filter
;
constexpr
auto
component_filter
=
string_view
{
""
};
extern
CAF_CORE_EXPORT
const
string_view
console
;
constexpr
auto
console
=
string_view
{
"none"
};
extern
CAF_CORE_EXPORT
string_view
console_format
;
constexpr
auto
console_format
=
string_view
{
"%m"
};
extern
CAF_CORE_EXPORT
const
string_view
console_verbosity
;
constexpr
auto
console_verbosity
=
default_log_level
;
extern
CAF_CORE_EXPORT
string_view
file_format
;
constexpr
auto
file_format
=
string_view
{
"%r %c %p %a %t %C %M %F:%L %m%n"
};
extern
CAF_CORE_EXPORT
string_view
file_name
;
constexpr
auto
file_verbosity
=
default_log_level
;
extern
CAF_CORE_EXPORT
const
string_view
file_verbosity
;
constexpr
auto
file_name
=
string_view
{
"actor_log_[PID]_[TIMESTAMP]_[NODE].log"
};
}
// namespace logger
}
// namespace
caf::defaults::
logger
namespace
middleman
{
namespace
caf
::
defaults
::
middleman
{
extern
CAF_CORE_EXPORT
std
::
vector
<
std
::
string
>
app_identifiers
;
constexpr
auto
app_identifier
=
string_view
{
"generic-caf-app"
};
extern
CAF_CORE_EXPORT
const
string_view
network_backend
;
constexpr
auto
network_backend
=
string_view
{
"default"
};
extern
CAF_CORE_EXPORT
const
size_t
max_consecutive_reads
;
constexpr
auto
max_consecutive_reads
=
size_t
{
50
};
extern
CAF_CORE_EXPORT
const
size_t
heartbeat_interval
;
constexpr
auto
heartbeat_interval
=
size_t
{
0
};
extern
CAF_CORE_EXPORT
const
size_t
cached_udp_buffers
;
constexpr
auto
cached_udp_buffers
=
size_t
{
10
};
extern
CAF_CORE_EXPORT
const
size_t
max_pending_msgs
;
constexpr
auto
max_pending_msgs
=
size_t
{
10
};
extern
CAF_CORE_EXPORT
const
size_t
workers
;
}
// namespace middleman
}
// namespace caf::defaults::middleman
}
// namespace caf::defaults
libcaf_core/caf/scheduler/abstract_coordinator.hpp
View file @
889ec7ef
...
@@ -85,6 +85,8 @@ public:
...
@@ -85,6 +85,8 @@ public:
virtual
actor_clock
&
clock
()
noexcept
=
0
;
virtual
actor_clock
&
clock
()
noexcept
=
0
;
static
size_t
default_thread_count
()
noexcept
;
protected:
protected:
void
stop_actors
();
void
stop_actors
();
...
...
libcaf_core/src/actor_system_config.cpp
View file @
889ec7ef
...
@@ -160,7 +160,6 @@ settings actor_system_config::dump_content() const {
...
@@ -160,7 +160,6 @@ settings actor_system_config::dump_content() const {
// -- scheduler parameters
// -- scheduler parameters
auto
&
scheduler_group
=
result
[
"scheduler"
].
as_dictionary
();
auto
&
scheduler_group
=
result
[
"scheduler"
].
as_dictionary
();
put_missing
(
scheduler_group
,
"policy"
,
defaults
::
scheduler
::
policy
);
put_missing
(
scheduler_group
,
"policy"
,
defaults
::
scheduler
::
policy
);
put_missing
(
scheduler_group
,
"max-threads"
,
defaults
::
scheduler
::
max_threads
);
put_missing
(
scheduler_group
,
"max-throughput"
,
put_missing
(
scheduler_group
,
"max-throughput"
,
defaults
::
scheduler
::
max_throughput
);
defaults
::
scheduler
::
max_throughput
);
put_missing
(
scheduler_group
,
"enable-profiling"
,
false
);
put_missing
(
scheduler_group
,
"enable-profiling"
,
false
);
...
@@ -196,14 +195,14 @@ settings actor_system_config::dump_content() const {
...
@@ -196,14 +195,14 @@ settings actor_system_config::dump_content() const {
put_missing
(
logger_group
,
"inline-output"
,
false
);
put_missing
(
logger_group
,
"inline-output"
,
false
);
// -- middleman parameters
// -- middleman parameters
auto
&
middleman_group
=
result
[
"middleman"
].
as_dictionary
();
auto
&
middleman_group
=
result
[
"middleman"
].
as_dictionary
();
auto
default_id
=
to_string
(
defaults
::
middleman
::
app_identifier
);
put_missing
(
middleman_group
,
"app-identifiers"
,
put_missing
(
middleman_group
,
"app-identifiers"
,
defaults
::
middleman
::
app_identifiers
);
std
::
vector
<
std
::
string
>
{
std
::
move
(
default_id
)}
);
put_missing
(
middleman_group
,
"enable-automatic-connections"
,
false
);
put_missing
(
middleman_group
,
"enable-automatic-connections"
,
false
);
put_missing
(
middleman_group
,
"max-consecutive-reads"
,
put_missing
(
middleman_group
,
"max-consecutive-reads"
,
defaults
::
middleman
::
max_consecutive_reads
);
defaults
::
middleman
::
max_consecutive_reads
);
put_missing
(
middleman_group
,
"heartbeat-interval"
,
put_missing
(
middleman_group
,
"heartbeat-interval"
,
defaults
::
middleman
::
heartbeat_interval
);
defaults
::
middleman
::
heartbeat_interval
);
put_missing
(
middleman_group
,
"workers"
,
defaults
::
middleman
::
workers
);
// -- openssl parameters
// -- openssl parameters
auto
&
openssl_group
=
result
[
"openssl"
].
as_dictionary
();
auto
&
openssl_group
=
result
[
"openssl"
].
as_dictionary
();
put_missing
(
openssl_group
,
"certificate"
,
std
::
string
{});
put_missing
(
openssl_group
,
"certificate"
,
std
::
string
{});
...
...
libcaf_core/src/defaults.cpp
deleted
100644 → 0
View file @
e74fd805
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/defaults.hpp"
#include <algorithm>
#include <chrono>
#include <limits>
#include <thread>
#include "caf/detail/build_config.hpp"
#include "caf/detail/log_level.hpp"
using
std
::
max
;
using
std
::
min
;
namespace
{
static
constexpr
caf
::
string_view
default_log_level
=
#if CAF_LOG_LEVEL == CAF_LOG_LEVEL_TRACE
"trace"
;
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_DEBUG
"debug"
;
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_INFO
"info"
;
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_WARNING
"warning"
;
#elif CAF_LOG_LEVEL == CAF_LOG_LEVEL_ERROR
"error"
;
#else
"quiet"
;
#endif
using
us_t
=
std
::
chrono
::
microseconds
;
constexpr
caf
::
timespan
us
(
us_t
::
rep
x
)
{
return
std
::
chrono
::
duration_cast
<
caf
::
timespan
>
(
us_t
{
x
});
}
using
ms_t
=
std
::
chrono
::
milliseconds
;
constexpr
caf
::
timespan
ms
(
ms_t
::
rep
x
)
{
return
std
::
chrono
::
duration_cast
<
caf
::
timespan
>
(
ms_t
{
x
});
}
}
// namespace
namespace
caf
::
defaults
{
namespace
stream
{
const
timespan
desired_batch_complexity
=
us
(
50
);
const
timespan
max_batch_delay
=
ms
(
5
);
const
timespan
credit_round_interval
=
ms
(
10
);
const
string_view
credit_policy
=
"complexity"
;
namespace
size_policy
{
const
int32_t
bytes_per_batch
=
2048
;
// 2 KB
const
int32_t
buffer_capacity
=
64
*
1024
;
// 64 KB
}
// namespace size_policy
}
// namespace stream
namespace
scheduler
{
const
string_view
policy
=
"stealing"
;
string_view
profiling_output_file
=
""
;
const
size_t
max_threads
=
max
(
std
::
thread
::
hardware_concurrency
(),
4u
);
const
size_t
max_throughput
=
std
::
numeric_limits
<
size_t
>::
max
();
const
timespan
profiling_resolution
=
ms
(
100
);
}
// namespace scheduler
namespace
work_stealing
{
const
size_t
aggressive_poll_attempts
=
100
;
const
size_t
aggressive_steal_interval
=
10
;
const
size_t
moderate_poll_attempts
=
500
;
const
size_t
moderate_steal_interval
=
5
;
const
timespan
moderate_sleep_duration
=
us
(
50
);
const
size_t
relaxed_steal_interval
=
1
;
const
timespan
relaxed_sleep_duration
=
ms
(
10
);
}
// namespace work_stealing
namespace
logger
{
string_view
component_filter
=
""
;
const
string_view
console
=
"none"
;
string_view
console_format
=
"%m"
;
const
string_view
console_verbosity
=
default_log_level
;
string_view
file_format
=
"%r %c %p %a %t %C %M %F:%L %m%n"
;
string_view
file_name
=
"actor_log_[PID]_[TIMESTAMP]_[NODE].log"
;
const
string_view
file_verbosity
=
default_log_level
;
}
// namespace logger
namespace
middleman
{
std
::
vector
<
std
::
string
>
app_identifiers
{
"generic-caf-app"
};
const
string_view
network_backend
=
"default"
;
const
size_t
max_consecutive_reads
=
50
;
const
size_t
heartbeat_interval
=
0
;
const
size_t
cached_udp_buffers
=
10
;
const
size_t
max_pending_msgs
=
10
;
const
size_t
workers
=
min
(
3u
,
std
::
thread
::
hardware_concurrency
()
/
4u
)
+
1
;
}
// namespace middleman
}
// namespace caf::defaults
libcaf_core/src/scheduler/abstract_coordinator.cpp
View file @
889ec7ef
...
@@ -18,6 +18,7 @@
...
@@ -18,6 +18,7 @@
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include <algorithm>
#include <atomic>
#include <atomic>
#include <chrono>
#include <chrono>
#include <condition_variable>
#include <condition_variable>
...
@@ -247,7 +248,10 @@ void abstract_coordinator::start() {
...
@@ -247,7 +248,10 @@ void abstract_coordinator::start() {
void
abstract_coordinator
::
init
(
actor_system_config
&
cfg
)
{
void
abstract_coordinator
::
init
(
actor_system_config
&
cfg
)
{
namespace
sr
=
defaults
::
scheduler
;
namespace
sr
=
defaults
::
scheduler
;
max_throughput_
=
get_or
(
cfg
,
"scheduler.max-throughput"
,
sr
::
max_throughput
);
max_throughput_
=
get_or
(
cfg
,
"scheduler.max-throughput"
,
sr
::
max_throughput
);
num_workers_
=
get_or
(
cfg
,
"scheduler.max-threads"
,
sr
::
max_threads
);
if
(
auto
num_workers
=
get_if
<
size_t
>
(
&
cfg
,
"scheduler.max-threads"
))
num_workers_
=
*
num_workers
;
else
num_workers_
=
default_thread_count
();
}
}
actor_system
::
module
::
id_t
abstract_coordinator
::
id
()
const
{
actor_system
::
module
::
id_t
abstract_coordinator
::
id
()
const
{
...
@@ -310,4 +314,8 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) {
...
@@ -310,4 +314,8 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) {
intrusive_ptr_release
(
ptr
);
intrusive_ptr_release
(
ptr
);
}
}
size_t
abstract_coordinator
::
default_thread_count
()
noexcept
{
return
std
::
max
(
std
::
thread
::
hardware_concurrency
(),
4u
);
}
}
// namespace caf::scheduler
}
// namespace caf::scheduler
libcaf_core/test/thread_hook.cpp
View file @
889ec7ef
...
@@ -115,8 +115,8 @@ CAF_TEST_FIXTURE_SCOPE(counting_hook, fixture<counting_thread_hook>)
...
@@ -115,8 +115,8 @@ CAF_TEST_FIXTURE_SCOPE(counting_hook, fixture<counting_thread_hook>)
CAF_TEST
(
counting_system_without_actor
)
{
CAF_TEST
(
counting_system_without_actor
)
{
assumed_init_calls
=
1
;
assumed_init_calls
=
1
;
a
ssumed_thread_count
=
get_or
(
cfg
,
"scheduler.max-threads"
,
a
uto
fallback
=
scheduler
::
abstract_coordinator
::
default_thread_count
();
defaults
::
scheduler
::
max_threads
)
assumed_thread_count
=
get_or
(
cfg
,
"scheduler.max-threads"
,
fallback
)
+
1
;
// caf.clock
+
1
;
// caf.clock
auto
&
sched
=
sys
.
scheduler
();
auto
&
sched
=
sys
.
scheduler
();
if
(
sched
.
detaches_utility_actors
())
if
(
sched
.
detaches_utility_actors
())
...
@@ -125,8 +125,8 @@ CAF_TEST(counting_system_without_actor) {
...
@@ -125,8 +125,8 @@ CAF_TEST(counting_system_without_actor) {
CAF_TEST
(
counting_system_with_actor
)
{
CAF_TEST
(
counting_system_with_actor
)
{
assumed_init_calls
=
1
;
assumed_init_calls
=
1
;
a
ssumed_thread_count
=
get_or
(
cfg
,
"scheduler.max-threads"
,
a
uto
fallback
=
scheduler
::
abstract_coordinator
::
default_thread_count
();
defaults
::
scheduler
::
max_threads
)
assumed_thread_count
=
get_or
(
cfg
,
"scheduler.max-threads"
,
fallback
)
+
2
;
// caf.clock and detached actor
+
2
;
// caf.clock and detached actor
auto
&
sched
=
sys
.
scheduler
();
auto
&
sched
=
sys
.
scheduler
();
if
(
sched
.
detaches_utility_actors
())
if
(
sched
.
detaches_utility_actors
())
...
...
libcaf_io/src/io/basp/instance.cpp
View file @
889ec7ef
...
@@ -18,6 +18,8 @@
...
@@ -18,6 +18,8 @@
#include "caf/io/basp/instance.hpp"
#include "caf/io/basp/instance.hpp"
#include <algorithm>
#include "caf/actor_system_config.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
...
@@ -41,8 +43,11 @@ instance::callee::~callee() {
...
@@ -41,8 +43,11 @@ instance::callee::~callee() {
instance
::
instance
(
abstract_broker
*
parent
,
callee
&
lstnr
)
instance
::
instance
(
abstract_broker
*
parent
,
callee
&
lstnr
)
:
tbl_
(
parent
),
this_node_
(
parent
->
system
().
node
()),
callee_
(
lstnr
)
{
:
tbl_
(
parent
),
this_node_
(
parent
->
system
().
node
()),
callee_
(
lstnr
)
{
CAF_ASSERT
(
this_node_
!=
none
);
CAF_ASSERT
(
this_node_
!=
none
);
auto
workers
size_t
workers
;
=
get_or
(
config
(),
"middleman.workers"
,
defaults
::
middleman
::
workers
);
if
(
auto
workers_cfg
=
get_if
<
size_t
>
(
&
config
(),
"middleman.workers"
))
workers
=
*
workers_cfg
;
else
workers
=
std
::
min
(
3u
,
std
::
thread
::
hardware_concurrency
()
/
4u
)
+
1
;
for
(
size_t
i
=
0
;
i
<
workers
;
++
i
)
for
(
size_t
i
=
0
;
i
<
workers
;
++
i
)
hub_
.
add_new_worker
(
queue_
,
proxies
());
hub_
.
add_new_worker
(
queue_
,
proxies
());
}
}
...
@@ -224,8 +229,12 @@ void instance::write_server_handshake(execution_unit* ctx, byte_buffer& out_buf,
...
@@ -224,8 +229,12 @@ void instance::write_server_handshake(execution_unit* ctx, byte_buffer& out_buf,
}
}
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
auto
writer
=
make_callback
([
&
](
binary_serializer
&
sink
)
{
auto
writer
=
make_callback
([
&
](
binary_serializer
&
sink
)
{
auto
app_ids
=
get_or
(
config
(),
"middleman.app-identifiers"
,
using
string_list
=
std
::
vector
<
std
::
string
>
;
defaults
::
middleman
::
app_identifiers
);
string_list
app_ids
;
if
(
auto
ids
=
get_if
<
string_list
>
(
&
config
(),
"middleman.app-identifiers"
))
app_ids
=
std
::
move
(
*
ids
);
else
app_ids
.
emplace_back
(
to_string
(
defaults
::
middleman
::
app_identifier
));
auto
aid
=
invalid_actor_id
;
auto
aid
=
invalid_actor_id
;
auto
iface
=
std
::
set
<
std
::
string
>
{};
auto
iface
=
std
::
set
<
std
::
string
>
{};
if
(
pa
!=
nullptr
&&
pa
->
first
!=
nullptr
)
{
if
(
pa
!=
nullptr
&&
pa
->
first
!=
nullptr
)
{
...
@@ -300,10 +309,11 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
...
@@ -300,10 +309,11 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
// Dispatch by message type.
// Dispatch by message type.
switch
(
hdr
.
operation
)
{
switch
(
hdr
.
operation
)
{
case
message_type
:
:
server_handshake
:
{
case
message_type
:
:
server_handshake
:
{
using
string_list
=
std
::
vector
<
std
::
string
>
;
// Deserialize payload.
// Deserialize payload.
binary_deserializer
bd
{
ctx
,
*
payload
};
binary_deserializer
bd
{
ctx
,
*
payload
};
node_id
source_node
;
node_id
source_node
;
st
d
::
vector
<
std
::
string
>
app_ids
;
st
ring_list
app_ids
;
actor_id
aid
=
invalid_actor_id
;
actor_id
aid
=
invalid_actor_id
;
std
::
set
<
std
::
string
>
sigs
;
std
::
set
<
std
::
string
>
sigs
;
if
(
auto
err
=
bd
(
source_node
,
app_ids
,
aid
,
sigs
))
{
if
(
auto
err
=
bd
(
source_node
,
app_ids
,
aid
,
sigs
))
{
...
@@ -312,8 +322,11 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
...
@@ -312,8 +322,11 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
return
serializing_basp_payload_failed
;
return
serializing_basp_payload_failed
;
}
}
// Check the application ID.
// Check the application ID.
auto
whitelist
=
get_or
(
config
(),
"middleman.app-identifiers"
,
string_list
whitelist
;
defaults
::
middleman
::
app_identifiers
);
if
(
auto
ls
=
get_if
<
string_list
>
(
&
config
(),
"middleman.app-identifiers"
))
whitelist
=
std
::
move
(
*
ls
);
else
whitelist
.
emplace_back
(
to_string
(
defaults
::
middleman
::
app_identifier
));
auto
i
=
std
::
find_first_of
(
app_ids
.
begin
(),
app_ids
.
end
(),
auto
i
=
std
::
find_first_of
(
app_ids
.
begin
(),
app_ids
.
end
(),
whitelist
.
begin
(),
whitelist
.
end
());
whitelist
.
begin
(),
whitelist
.
end
());
if
(
i
==
app_ids
.
end
())
{
if
(
i
==
app_ids
.
end
())
{
...
...
libcaf_io/test/io/basp_broker.cpp
View file @
889ec7ef
...
@@ -109,6 +109,7 @@ public:
...
@@ -109,6 +109,7 @@ public:
.
set
(
"middleman.workers"
,
size_t
{
0
})
.
set
(
"middleman.workers"
,
size_t
{
0
})
.
set
(
"scheduler.policy"
,
autoconn
?
"testing"
:
"stealing"
)
.
set
(
"scheduler.policy"
,
autoconn
?
"testing"
:
"stealing"
)
.
set
(
"middleman.attach-utility-actors"
,
autoconn
))
{
.
set
(
"middleman.attach-utility-actors"
,
autoconn
))
{
app_ids
.
emplace_back
(
to_string
(
defaults
::
middleman
::
app_identifier
));
auto
&
mm
=
sys
.
middleman
();
auto
&
mm
=
sys
.
middleman
();
mpx_
=
dynamic_cast
<
network
::
test_multiplexer
*>
(
&
mm
.
backend
());
mpx_
=
dynamic_cast
<
network
::
test_multiplexer
*>
(
&
mm
.
backend
());
CAF_REQUIRE
(
mpx_
!=
nullptr
);
CAF_REQUIRE
(
mpx_
!=
nullptr
);
...
@@ -268,8 +269,7 @@ public:
...
@@ -268,8 +269,7 @@ public:
n
.
id
)
n
.
id
)
.
receive
(
hdl
,
basp
::
message_type
::
server_handshake
,
no_flags
,
any_vals
,
.
receive
(
hdl
,
basp
::
message_type
::
server_handshake
,
no_flags
,
any_vals
,
basp
::
version
,
invalid_actor_id
,
invalid_actor_id
,
this_node
(),
basp
::
version
,
invalid_actor_id
,
invalid_actor_id
,
this_node
(),
defaults
::
middleman
::
app_identifiers
,
published_actor_id
,
app_ids
,
published_actor_id
,
published_actor_ifs
)
published_actor_ifs
)
// upon receiving our client handshake, BASP will check
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
// whether there is a SpawnServ actor on this node
.
receive
(
hdl
,
basp
::
message_type
::
direct_message
,
.
receive
(
hdl
,
basp
::
message_type
::
direct_message
,
...
@@ -396,6 +396,7 @@ public:
...
@@ -396,6 +396,7 @@ public:
actor_system_config
cfg
;
actor_system_config
cfg
;
actor_system
sys
;
actor_system
sys
;
std
::
vector
<
std
::
string
>
app_ids
;
private:
private:
basp_broker
*
aut_
;
basp_broker
*
aut_
;
...
@@ -483,8 +484,7 @@ CAF_TEST(non_empty_server_handshake) {
...
@@ -483,8 +484,7 @@ CAF_TEST(non_empty_server_handshake) {
byte_buffer
expected_payload
;
byte_buffer
expected_payload
;
std
::
set
<
std
::
string
>
ifs
{
"caf::replies_to<@u16>::with<@u16>"
};
std
::
set
<
std
::
string
>
ifs
{
"caf::replies_to<@u16>::with<@u16>"
};
binary_serializer
sink
{
nullptr
,
expected_payload
};
binary_serializer
sink
{
nullptr
,
expected_payload
};
if
(
auto
err
=
sink
(
instance
().
this_node
(),
if
(
auto
err
=
sink
(
instance
().
this_node
(),
app_ids
,
self
()
->
id
(),
ifs
))
defaults
::
middleman
::
app_identifiers
,
self
()
->
id
(),
ifs
))
CAF_FAIL
(
"serializing handshake failed: "
<<
sys
.
render
(
err
));
CAF_FAIL
(
"serializing handshake failed: "
<<
sys
.
render
(
err
));
CAF_CHECK_EQUAL
(
hexstr
(
payload
),
hexstr
(
expected_payload
));
CAF_CHECK_EQUAL
(
hexstr
(
payload
),
hexstr
(
expected_payload
));
}
}
...
@@ -579,8 +579,8 @@ CAF_TEST(remote_actor_and_send) {
...
@@ -579,8 +579,8 @@ CAF_TEST(remote_actor_and_send) {
mock
(
jupiter
().
connection
,
mock
(
jupiter
().
connection
,
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
invalid_actor_id
,
invalid_actor_id
},
invalid_actor_id
,
invalid_actor_id
},
jupiter
().
id
,
defaults
::
middleman
::
app_identifiers
,
jupiter
().
id
,
app_ids
,
jupiter
().
dummy_actor
->
id
()
,
jupiter
().
dummy_actor
->
id
(),
std
::
set
<
std
::
string
>
{})
std
::
set
<
std
::
string
>
{})
.
receive
(
jupiter
().
connection
,
basp
::
message_type
::
client_handshake
,
.
receive
(
jupiter
().
connection
,
basp
::
message_type
::
client_handshake
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
invalid_actor_id
,
this_node
())
invalid_actor_id
,
this_node
())
...
@@ -773,8 +773,8 @@ CAF_TEST(automatic_connection) {
...
@@ -773,8 +773,8 @@ CAF_TEST(automatic_connection) {
mock
(
jupiter
().
connection
,
mock
(
jupiter
().
connection
,
{
basp
::
message_type
::
server_handshake
,
no_flags
,
0
,
basp
::
version
,
{
basp
::
message_type
::
server_handshake
,
no_flags
,
0
,
basp
::
version
,
invalid_actor_id
,
invalid_actor_id
},
invalid_actor_id
,
invalid_actor_id
},
jupiter
().
id
,
defaults
::
middleman
::
app_identifiers
,
jupiter
().
id
,
app_ids
,
jupiter
().
dummy_actor
->
id
()
,
jupiter
().
dummy_actor
->
id
(),
std
::
set
<
std
::
string
>
{})
std
::
set
<
std
::
string
>
{})
.
receive
(
jupiter
().
connection
,
basp
::
message_type
::
client_handshake
,
.
receive
(
jupiter
().
connection
,
basp
::
message_type
::
client_handshake
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
invalid_actor_id
,
this_node
());
invalid_actor_id
,
this_node
());
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment