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
03b33445
Commit
03b33445
authored
Jan 19, 2017
by
Dominik Charousset
Committed by
Marian Triebe
Apr 23, 2017
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Apply clang-tidy to codebase
parent
ece584cc
Changes
16
Hide whitespace changes
Inline
Side-by-side
Showing
16 changed files
with
45 additions
and
40 deletions
+45
-40
libcaf_core/caf/abstract_upstream.hpp
libcaf_core/caf/abstract_upstream.hpp
+1
-1
libcaf_core/caf/stream_handler.hpp
libcaf_core/caf/stream_handler.hpp
+1
-1
libcaf_core/caf/stream_msg.hpp
libcaf_core/caf/stream_msg.hpp
+3
-2
libcaf_core/caf/stream_msg_visitor.hpp
libcaf_core/caf/stream_msg_visitor.hpp
+1
-1
libcaf_core/caf/stream_sink.hpp
libcaf_core/caf/stream_sink.hpp
+1
-1
libcaf_core/caf/stream_source.hpp
libcaf_core/caf/stream_source.hpp
+1
-1
libcaf_core/caf/upstream_path.hpp
libcaf_core/caf/upstream_path.hpp
+1
-1
libcaf_core/src/abstract_downstream.cpp
libcaf_core/src/abstract_downstream.cpp
+5
-1
libcaf_core/src/actor_system_config.cpp
libcaf_core/src/actor_system_config.cpp
+2
-2
libcaf_core/src/logger.cpp
libcaf_core/src/logger.cpp
+2
-2
libcaf_core/src/stream_msg_visitor.cpp
libcaf_core/src/stream_msg_visitor.cpp
+16
-18
libcaf_core/src/stream_source.cpp
libcaf_core/src/stream_source.cpp
+4
-4
libcaf_core/src/stream_stage.cpp
libcaf_core/src/stream_stage.cpp
+1
-1
libcaf_core/src/term.cpp
libcaf_core/src/term.cpp
+1
-1
libcaf_core/src/upstream_path.cpp
libcaf_core/src/upstream_path.cpp
+4
-2
libcaf_core/test/streaming.cpp
libcaf_core/test/streaming.cpp
+1
-1
No files found.
libcaf_core/caf/abstract_upstream.hpp
View file @
03b33445
...
@@ -51,7 +51,7 @@ public:
...
@@ -51,7 +51,7 @@ public:
/// @pre `self != nullptr`
/// @pre `self != nullptr`
/// @pre `policy != nullptr`
/// @pre `policy != nullptr`
abstract_upstream
(
local_actor
*
self
,
policy_ptr
policy
);
abstract_upstream
(
local_actor
*
self
ptr
,
policy_ptr
p
);
virtual
~
abstract_upstream
();
virtual
~
abstract_upstream
();
...
...
libcaf_core/caf/stream_handler.hpp
View file @
03b33445
...
@@ -32,7 +32,7 @@ namespace caf {
...
@@ -32,7 +32,7 @@ namespace caf {
/// Manages a single stream with any number of down- and upstream actors.
/// Manages a single stream with any number of down- and upstream actors.
class
stream_handler
:
public
ref_counted
{
class
stream_handler
:
public
ref_counted
{
public:
public:
~
stream_handler
();
~
stream_handler
()
override
;
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
...
...
libcaf_core/caf/stream_msg.hpp
View file @
03b33445
...
@@ -20,6 +20,7 @@
...
@@ -20,6 +20,7 @@
#ifndef CAF_STREAM_MSG_HPP
#ifndef CAF_STREAM_MSG_HPP
#define CAF_STREAM_MSG_HPP
#define CAF_STREAM_MSG_HPP
#include <utility>
#include <vector>
#include <vector>
#include <cstdint>
#include <cstdint>
...
@@ -145,8 +146,8 @@ struct stream_msg : tag::boxing_type {
...
@@ -145,8 +146,8 @@ struct stream_msg : tag::boxing_type {
content_type
content
;
content_type
content
;
template
<
class
T
>
template
<
class
T
>
stream_msg
(
const
stream_id
&
id
,
T
&&
x
)
stream_msg
(
stream_id
id
,
T
&&
x
)
:
sid
(
id
),
:
sid
(
std
::
move
(
id
)
),
content
(
std
::
forward
<
T
>
(
x
))
{
content
(
std
::
forward
<
T
>
(
x
))
{
// nop
// nop
}
}
...
...
libcaf_core/caf/stream_msg_visitor.hpp
View file @
03b33445
...
@@ -37,7 +37,7 @@ public:
...
@@ -37,7 +37,7 @@ public:
using
result_type
=
std
::
pair
<
error
,
iterator
>
;
using
result_type
=
std
::
pair
<
error
,
iterator
>
;
stream_msg_visitor
(
scheduled_actor
*
self
,
stream_id
&
sid
,
stream_msg_visitor
(
scheduled_actor
*
self
,
stream_id
&
sid
,
iterator
pos
,
iterator
end
);
iterator
i
,
iterator
last
);
result_type
operator
()(
stream_msg
::
open
&
x
);
result_type
operator
()(
stream_msg
::
open
&
x
);
...
...
libcaf_core/caf/stream_sink.hpp
View file @
03b33445
...
@@ -39,7 +39,7 @@ public:
...
@@ -39,7 +39,7 @@ public:
stream_sink
(
abstract_upstream
*
in_ptr
,
strong_actor_ptr
&&
orig_sender
,
stream_sink
(
abstract_upstream
*
in_ptr
,
strong_actor_ptr
&&
orig_sender
,
std
::
vector
<
strong_actor_ptr
>&&
trailing_stages
,
message_id
mid
);
std
::
vector
<
strong_actor_ptr
>&&
trailing_stages
,
message_id
mid
);
bool
done
()
const
;
bool
done
()
const
override
;
error
upstream_batch
(
strong_actor_ptr
&
src
,
size_t
xs_size
,
error
upstream_batch
(
strong_actor_ptr
&
src
,
size_t
xs_size
,
message
&
xs
)
final
;
message
&
xs
)
final
;
...
...
libcaf_core/caf/stream_source.hpp
View file @
03b33445
...
@@ -35,7 +35,7 @@ class stream_source : public extend<stream_handler, stream_source>::
...
@@ -35,7 +35,7 @@ class stream_source : public extend<stream_handler, stream_source>::
public:
public:
stream_source
(
abstract_downstream
*
out_ptr
);
stream_source
(
abstract_downstream
*
out_ptr
);
~
stream_source
();
~
stream_source
()
override
;
bool
done
()
const
final
;
bool
done
()
const
final
;
...
...
libcaf_core/caf/upstream_path.hpp
View file @
03b33445
...
@@ -51,7 +51,7 @@ public:
...
@@ -51,7 +51,7 @@ public:
/// Amount of credit we have signaled upstream.
/// Amount of credit we have signaled upstream.
size_t
assigned_credit
;
size_t
assigned_credit
;
upstream_path
(
strong_actor_ptr
hdl
,
const
stream_id
&
id
,
stream_priority
p
);
upstream_path
(
strong_actor_ptr
ptr
,
stream_id
id
,
stream_priority
p
);
};
};
}
// namespace caf
}
// namespace caf
...
...
libcaf_core/src/abstract_downstream.cpp
View file @
03b33445
...
@@ -17,6 +17,8 @@
...
@@ -17,6 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
******************************************************************************/
#include <utility>
#include "caf/abstract_downstream.hpp"
#include "caf/abstract_downstream.hpp"
#include "caf/send.hpp"
#include "caf/send.hpp"
...
@@ -28,7 +30,9 @@ namespace caf {
...
@@ -28,7 +30,9 @@ namespace caf {
abstract_downstream
::
abstract_downstream
(
local_actor
*
selfptr
,
abstract_downstream
::
abstract_downstream
(
local_actor
*
selfptr
,
const
stream_id
&
sid
,
const
stream_id
&
sid
,
std
::
unique_ptr
<
downstream_policy
>
ptr
)
std
::
unique_ptr
<
downstream_policy
>
ptr
)
:
self_
(
selfptr
),
sid_
(
sid
),
policy_
(
std
::
move
(
ptr
))
{
:
self_
(
selfptr
),
sid_
(
sid
),
policy_
(
std
::
move
(
ptr
))
{
// nop
// nop
}
}
...
...
libcaf_core/src/actor_system_config.cpp
View file @
03b33445
...
@@ -71,7 +71,7 @@ public:
...
@@ -71,7 +71,7 @@ public:
}
}
if
(
out
)
if
(
out
)
*
out
<<
"error in line "
<<
ln
*
out
<<
"error in line "
<<
ln
<<
": unrecognized parameter name
\"
"
<<
name
<<
"
\"
"
<<
R"(: unrecognized parameter name ")"
<<
name
<<
R"(")
"
<<
std
::
endl
;
<<
std
::
endl
;
}
}
...
@@ -271,7 +271,7 @@ actor_system_config& actor_system_config::parse(message& args,
...
@@ -271,7 +271,7 @@ actor_system_config& actor_system_config::parse(message& args,
i
->
second
(
ln
,
cv
,
none
);
i
->
second
(
ln
,
cv
,
none
);
else
else
std
::
cerr
<<
"error in line "
<<
ln
std
::
cerr
<<
"error in line "
<<
ln
<<
": unrecognized parameter name
\"
"
<<
nm
<<
"
\"
"
<<
R"(: unrecognized parameter name ")"
<<
nm
<<
R"(")
"
<<
std
::
endl
;
<<
std
::
endl
;
};
};
actor_system_config_reader
consumer
{
options_
,
custom_options_
,
nac_sink
};
actor_system_config_reader
consumer
{
options_
,
custom_options_
,
nac_sink
};
...
...
libcaf_core/src/logger.cpp
View file @
03b33445
...
@@ -80,7 +80,7 @@ pthread_key_t s_key;
...
@@ -80,7 +80,7 @@ pthread_key_t s_key;
pthread_once_t
s_key_once
=
PTHREAD_ONCE_INIT
;
pthread_once_t
s_key_once
=
PTHREAD_ONCE_INIT
;
void
logger_ptr_destructor
(
void
*
ptr
)
{
void
logger_ptr_destructor
(
void
*
ptr
)
{
if
(
ptr
)
{
if
(
ptr
!=
nullptr
)
{
intrusive_ptr_release
(
reinterpret_cast
<
logger
*>
(
ptr
));
intrusive_ptr_release
(
reinterpret_cast
<
logger
*>
(
ptr
));
}
}
}
}
...
@@ -92,7 +92,7 @@ void make_logger_ptr() {
...
@@ -92,7 +92,7 @@ void make_logger_ptr() {
void
set_current_logger
(
logger
*
x
)
{
void
set_current_logger
(
logger
*
x
)
{
pthread_once
(
&
s_key_once
,
make_logger_ptr
);
pthread_once
(
&
s_key_once
,
make_logger_ptr
);
logger_ptr_destructor
(
pthread_getspecific
(
s_key
));
logger_ptr_destructor
(
pthread_getspecific
(
s_key
));
if
(
x
)
if
(
x
!=
nullptr
)
intrusive_ptr_add_ref
(
x
);
intrusive_ptr_add_ref
(
x
);
pthread_setspecific
(
s_key
,
x
);
pthread_setspecific
(
s_key
,
x
);
}
}
...
...
libcaf_core/src/stream_msg_visitor.cpp
View file @
03b33445
...
@@ -68,25 +68,23 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
...
@@ -68,25 +68,23 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
"handler after receiving token:"
"handler after receiving token:"
<<
CAF_ARG
(
x
.
token
));
<<
CAF_ARG
(
x
.
token
));
return
fail
(
sec
::
stream_init_failed
);
return
fail
(
sec
::
stream_init_failed
);
}
else
{
auto
&
handler
=
i_
->
second
;
// store upstream actor
auto
initial_credit
=
handler
->
add_upstream
(
x
.
prev_stage
,
sid_
,
x
.
priority
);
if
(
initial_credit
)
{
// check whether we are a stage in a longer pipeline and send more
// stream_open messages if required
auto
ic
=
static_cast
<
int32_t
>
(
*
initial_credit
);
std
::
vector
<
atom_value
>
filter
;
unsafe_send_as
(
self_
,
predecessor
,
make_message
(
make
<
stream_msg
::
ack_open
>
(
std
::
move
(
sid_
),
ic
,
std
::
move
(
filter
),
false
)));
return
{
none
,
i_
};
}
else
{
printf
(
"error: %s
\n
"
,
self_
->
system
().
render
(
initial_credit
.
error
()).
c_str
());
self_
->
streams
().
erase
(
i_
);
return
fail
(
std
::
move
(
initial_credit
.
error
()));
}
}
}
auto
&
handler
=
i_
->
second
;
// store upstream actor
auto
initial_credit
=
handler
->
add_upstream
(
x
.
prev_stage
,
sid_
,
x
.
priority
);
if
(
initial_credit
)
{
// check whether we are a stage in a longer pipeline and send more
// stream_open messages if required
auto
ic
=
static_cast
<
int32_t
>
(
*
initial_credit
);
std
::
vector
<
atom_value
>
filter
;
unsafe_send_as
(
self_
,
predecessor
,
make_message
(
make
<
stream_msg
::
ack_open
>
(
std
::
move
(
sid_
),
ic
,
std
::
move
(
filter
),
false
)));
return
{
none
,
i_
};
}
printf
(
"error: %s
\n
"
,
self_
->
system
().
render
(
initial_credit
.
error
()).
c_str
());
self_
->
streams
().
erase
(
i_
);
return
fail
(
std
::
move
(
initial_credit
.
error
()));
}
}
auto
stream_msg_visitor
::
operator
()(
stream_msg
::
close
&
)
->
result_type
{
auto
stream_msg_visitor
::
operator
()(
stream_msg
::
close
&
)
->
result_type
{
...
...
libcaf_core/src/stream_source.cpp
View file @
03b33445
...
@@ -52,12 +52,12 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
...
@@ -52,12 +52,12 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
if
(
current_size
<
size_hint
)
if
(
current_size
<
size_hint
)
generate
(
size_hint
-
current_size
);
generate
(
size_hint
-
current_size
);
return
trigger_send
(
hdl
);
return
trigger_send
(
hdl
);
}
else
if
(
buf_size
()
>
0
)
{
}
// transmit cached elements before closing paths
// transmit cached elements before closing paths
if
(
buf_size
()
>
0
)
return
trigger_send
(
hdl
);
return
trigger_send
(
hdl
);
}
else
if
(
out_ptr_
->
remove_path
(
hdl
))
{
if
(
out_ptr_
->
remove_path
(
hdl
))
return
none
;
return
none
;
}
}
}
return
sec
::
invalid_downstream
;
return
sec
::
invalid_downstream
;
}
}
...
...
libcaf_core/src/stream_stage.cpp
View file @
03b33445
...
@@ -56,7 +56,7 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
...
@@ -56,7 +56,7 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
path
->
open_credit
+=
value
;
path
->
open_credit
+=
value
;
if
(
out_ptr_
->
buf_size
()
>
0
)
{
if
(
out_ptr_
->
buf_size
()
>
0
)
{
return
trigger_send
(
hdl
);
return
trigger_send
(
hdl
);
}
else
if
(
in_ptr_
->
closed
())
{
}
if
(
in_ptr_
->
closed
())
{
if
(
!
out_ptr_
->
remove_path
(
hdl
))
{
if
(
!
out_ptr_
->
remove_path
(
hdl
))
{
return
sec
::
invalid_downstream
;
return
sec
::
invalid_downstream
;
}
}
...
...
libcaf_core/src/term.cpp
View file @
03b33445
...
@@ -27,7 +27,7 @@
...
@@ -27,7 +27,7 @@
#include <io.h>
#include <io.h>
#include <windows.h>
#include <windows.h>
#else
#else
#include <
stdio.h
>
#include <
cstdio
>
#include <unistd.h>
#include <unistd.h>
#endif
#endif
...
...
libcaf_core/src/upstream_path.cpp
View file @
03b33445
...
@@ -17,14 +17,16 @@
...
@@ -17,14 +17,16 @@
* http://www.boost.org/LICENSE_1_0.txt. *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
******************************************************************************/
#include <utility>
#include "caf/upstream_path.hpp"
#include "caf/upstream_path.hpp"
namespace
caf
{
namespace
caf
{
upstream_path
::
upstream_path
(
strong_actor_ptr
ptr
,
const
stream_id
&
id
,
upstream_path
::
upstream_path
(
strong_actor_ptr
ptr
,
stream_id
id
,
stream_priority
p
)
stream_priority
p
)
:
hdl
(
std
::
move
(
ptr
)),
:
hdl
(
std
::
move
(
ptr
)),
sid
(
id
),
sid
(
std
::
move
(
id
)
),
prio
(
p
),
prio
(
p
),
last_acked_batch_id
(
0
),
last_acked_batch_id
(
0
),
last_batch_id
(
0
),
last_batch_id
(
0
),
...
...
libcaf_core/test/streaming.cpp
View file @
03b33445
...
@@ -98,7 +98,7 @@ behavior filter(event_based_actor* self) {
...
@@ -98,7 +98,7 @@ behavior filter(event_based_actor* self) {
},
},
// processing step
// processing step
[
=
](
unit_t
&
,
downstream
<
int
>&
out
,
int
x
)
{
[
=
](
unit_t
&
,
downstream
<
int
>&
out
,
int
x
)
{
if
(
x
&
0x01
)
if
(
(
x
&
0x01
)
!=
0
)
out
.
push
(
x
);
out
.
push
(
x
);
},
},
// cleanup
// cleanup
...
...
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