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
7e05665f
Commit
7e05665f
authored
Jul 31, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Divide scribe_policy into source and header file. Also Fix some bugs
parent
73f52265
Changes
3
Hide whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
98 additions
and
53 deletions
+98
-53
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/policy/scribe_policy.hpp
libcaf_net/caf/policy/scribe_policy.hpp
+30
-53
libcaf_net/src/scribe_policy.cpp
libcaf_net/src/scribe_policy.cpp
+67
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
7e05665f
...
...
@@ -18,6 +18,7 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe_policy.cpp
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/policy/scribe_policy.hpp
View file @
7e05665f
...
...
@@ -25,6 +25,7 @@
#include <caf/logger.hpp>
#include <caf/sec.hpp>
#include <caf/variant.hpp>
#include <caf/net/endpoint_manager.hpp>
#ifdef CAF_WINDOWS
# include <winsock2.h>
...
...
@@ -41,7 +42,14 @@ namespace policy {
class
scribe_policy
{
public:
scribe_policy
(
net
::
stream_socket
handle
)
:
handle_
(
handle
)
{
explicit
scribe_policy
(
net
::
stream_socket
handle
)
:
handle_
(
handle
),
max_consecutive_reads_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
)
{
// nop
}
...
...
@@ -52,7 +60,9 @@ public:
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
error
init
(
Parent
&
parent
)
{
prepare_next_read
();
parent
.
mask_add
(
net
::
operation
::
read_write
);
return
none
;
}
...
...
@@ -64,6 +74,7 @@ public:
auto
rres
=
read
(
handle_
,
buf
,
len
);
if
(
rres
.
is
<
caf
::
sec
>
())
{
// Make sure WSAGetLastError gets called immediately on Windows.
CAF_LOG_DEBUG
(
"receive failed"
<<
CAF_ARG
(
get
<
sec
>
(
rres
)));
handle_error
(
parent
,
get
<
caf
::
sec
>
(
rres
));
return
false
;
}
...
...
@@ -73,7 +84,8 @@ public:
static_cast
<
size_t
>
(
get
<
size_t
>
(
rres
))
:
0
;
collected_
+=
result
;
if
(
collected_
>=
read_threshold_
)
{
collected_
=
0
;
parent
.
application
().
process
(
read_buf_
,
*
this
,
parent
);
prepare_next_read
();
return
false
;
}
else
{
return
true
;
...
...
@@ -82,16 +94,12 @@ public:
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
parent
)
{
if
(
!
write_buf_
.
empty
())
while
(
write_some
(
parent
));
// write while write_buf not empty
// check for new messages in parents message_queue
auto
&
msg_queue
=
parent
.
message_queue
();
while
(
!
msg_queue
.
empty
())
{
// TODO: would be nice to have some kind of `get_next_elem()` function
auto
&
elem
=
msg_queue
.
dequeue
();
// TODO parameter list from proposal -> pass to application
parent
->
application
().
prepare
(
elem
,
*
this
,
parent
);
while
(
write_some
(
parent
));
// write while write_buf not empty
// check new messages in parents message_queue
std
::
unique_ptr
<
caf
::
net
::
endpoint_manager
::
message
>
msg
;
while
((
msg
=
parent
.
next_message
()))
{
parent
.
application
().
prepare
(
std
::
move
(
msg
),
*
this
,
parent
);
}
// write prepared data
return
write_some
(
parent
);
...
...
@@ -100,6 +108,7 @@ public:
template
<
class
Parent
>
bool
write_some
(
Parent
&
parent
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
fd
)
<<
CAF_ARG
(
len
));
if
(
write_buf_
.
empty
())
return
false
;
auto
len
=
write_buf_
.
size
()
-
written_
;
void
*
buf
=
write_buf_
.
data
()
+
written_
;
auto
sres
=
net
::
write
(
handle_
,
buf
,
len
);
...
...
@@ -124,61 +133,29 @@ public:
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
parent
->
application
().
resolve
(
*
this
,
path
,
listener
);
parent
.
application
().
resolve
(
*
this
,
path
,
listener
);
// TODO should parent be passed as well?
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
parent
->
application
().
timeout
(
*
this
,
value
,
id
);
parent
.
application
().
timeout
(
*
this
,
value
,
id
);
// TODO should parent be passed as well?
}
template
<
class
Parent
>
void
handle_error
(
Parent
&
parent
,
sec
code
)
{
parent
.
application
()
.
handle_error
(
code
);
template
<
class
Application
>
void
handle_error
(
Application
&
application
,
sec
code
)
{
application
.
handle_error
(
code
);
}
void
prepare_next_read
()
{
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
void
prepare_next_read
();
void
configure_read
(
net
::
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
std
::
vector
<
char
>&
wr_buf
()
{
return
write_buf_
;
}
void
configure_read
(
net
::
receive_policy
::
config
cfg
);
std
::
vector
<
char
>&
wr_buf
();
private:
std
::
vector
<
char
>
read_buf_
;
std
::
vector
<
char
>
write_buf_
;
size_t
max_consecutive_reads_
;
...
...
libcaf_net/src/scribe_policy.cpp
0 → 100644
View file @
7e05665f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/policy/scribe_policy.hpp"
#include <system_error>
#include "caf/config.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
policy
{
void
scribe_policy
::
prepare_next_read
()
{
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
void
scribe_policy
::
configure_read
(
net
::
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
std
::
vector
<
char
>&
scribe_policy
::
wr_buf
()
{
return
write_buf_
;
}
}
// namespace policy
}
// namespace caf
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