Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
actor-incubator
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
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-incubator
Commits
3320d23e
Commit
3320d23e
authored
Oct 16, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Port message_queue from io
parent
29535532
Changes
4
Show whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
314 additions
and
0 deletions
+314
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/net/basp/message_queue.hpp
libcaf_net/caf/net/basp/message_queue.hpp
+81
-0
libcaf_net/src/message_queue.cpp
libcaf_net/src/message_queue.cpp
+77
-0
libcaf_net/test/net/basp/message_queue.cpp
libcaf_net/test/net/basp/message_queue.cpp
+155
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
3320d23e
...
@@ -33,6 +33,7 @@ set(LIBCAF_NET_SRCS
...
@@ -33,6 +33,7 @@ set(LIBCAF_NET_SRCS
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
src/udp_datagram_socket.cpp
src/defaults.cpp
src/defaults.cpp
src/message_queue.cpp
)
)
add_custom_target
(
libcaf_net
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/net/basp/message_queue.hpp
0 → 100644
View file @
3320d23e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <mutex>
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
namespace
caf
{
namespace
net
{
namespace
basp
{
/// Enforces strict order of message delivery, i.e., deliver messages in the
/// same order as if they were deserialized by a single thread.
class
message_queue
{
public:
// -- member types -----------------------------------------------------------
/// Request for sending a message to an actor at a later time.
struct
actor_msg
{
uint64_t
id
;
strong_actor_ptr
receiver
;
mailbox_element_ptr
content
;
};
// -- constructors, destructors, and assignment operators --------------------
message_queue
();
// -- mutators ---------------------------------------------------------------
/// Adds a new message to the queue or deliver it immediately if possible.
void
push
(
execution_unit
*
ctx
,
uint64_t
id
,
strong_actor_ptr
receiver
,
mailbox_element_ptr
content
);
/// Marks given ID as dropped, effectively skipping it without effect.
void
drop
(
execution_unit
*
ctx
,
uint64_t
id
);
/// Returns the next ascending ID.
uint64_t
new_id
();
// -- member variables -------------------------------------------------------
/// Protects all other properties.
std
::
mutex
lock
;
/// The next available ascending ID. The counter is large enough to overflow
/// after roughly 600 years if we dispatch a message every microsecond.
uint64_t
next_id
;
/// The next ID that we can ship.
uint64_t
next_undelivered
;
/// Keeps messages in sorted order in case a message other than
/// `next_undelivered` gets ready first.
std
::
vector
<
actor_msg
>
pending
;
};
}
// namespace basp
}
// namespace io
}
// namespace caf
libcaf_net/src/message_queue.cpp
0 → 100644
View file @
3320d23e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/net/basp/message_queue.hpp"
#include <iterator>
namespace
caf
{
namespace
net
{
namespace
basp
{
message_queue
::
message_queue
()
:
next_id
(
0
),
next_undelivered
(
0
)
{
// nop
}
void
message_queue
::
push
(
execution_unit
*
ctx
,
uint64_t
id
,
strong_actor_ptr
receiver
,
mailbox_element_ptr
content
)
{
std
::
unique_lock
<
std
::
mutex
>
guard
{
lock
};
CAF_ASSERT
(
id
>=
next_undelivered
);
CAF_ASSERT
(
id
<
next_id
);
auto
first
=
pending
.
begin
();
auto
last
=
pending
.
end
();
if
(
id
==
next_undelivered
)
{
// Dispatch current head.
if
(
receiver
!=
nullptr
)
receiver
->
enqueue
(
std
::
move
(
content
),
ctx
);
auto
next
=
id
+
1
;
// Check whether we can deliver more.
if
(
first
==
last
||
first
->
id
!=
next
)
{
next_undelivered
=
next
;
CAF_ASSERT
(
next_undelivered
<=
next_id
);
return
;
}
// Deliver everything until reaching a non-consecutive ID or the end.
auto
i
=
first
;
for
(;
i
!=
last
&&
i
->
id
==
next
;
++
i
,
++
next
)
if
(
i
->
receiver
!=
nullptr
)
i
->
receiver
->
enqueue
(
std
::
move
(
i
->
content
),
ctx
);
next_undelivered
=
next
;
pending
.
erase
(
first
,
i
);
CAF_ASSERT
(
next_undelivered
<=
next_id
);
return
;
}
// Get the insertion point.
auto
pred
=
[
&
](
const
actor_msg
&
x
)
{
return
x
.
id
>=
id
;
};
pending
.
emplace
(
std
::
find_if
(
first
,
last
,
pred
),
actor_msg
{
id
,
std
::
move
(
receiver
),
std
::
move
(
content
)});
}
void
message_queue
::
drop
(
execution_unit
*
ctx
,
uint64_t
id
)
{
push
(
ctx
,
id
,
nullptr
,
nullptr
);
}
uint64_t
message_queue
::
new_id
()
{
std
::
unique_lock
<
std
::
mutex
>
guard
{
lock
};
return
next_id
++
;
}
}
// namespace basp
}
// namespace io
}
// namespace caf
libcaf_net/test/net/basp/message_queue.cpp
0 → 100644
View file @
3320d23e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE basp.message_queue
#include "caf/net/basp/message_queue.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/behavior.hpp"
using
namespace
caf
;
namespace
{
behavior
testee_impl
()
{
return
{[](
ok_atom
,
int
)
{
// nop
}};
}
struct
fixture
:
test_coordinator_fixture
<>
{
net
::
basp
::
message_queue
queue
;
strong_actor_ptr
testee
;
fixture
()
{
auto
hdl
=
sys
.
spawn
<
lazy_init
>
(
testee_impl
);
testee
=
actor_cast
<
strong_actor_ptr
>
(
hdl
);
}
void
acquire_ids
(
size_t
num
)
{
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
queue
.
new_id
();
}
void
push
(
int
msg_id
)
{
queue
.
push
(
nullptr
,
static_cast
<
uint64_t
>
(
msg_id
),
testee
,
make_mailbox_element
(
self
->
ctrl
(),
make_message_id
(),
{},
ok_atom
::
value
,
msg_id
));
}
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
message_queue_tests
,
fixture
)
CAF_TEST
(
default
construction
)
{
CAF_CHECK_EQUAL
(
queue
.
next_id
,
0u
);
CAF_CHECK_EQUAL
(
queue
.
next_undelivered
,
0u
);
CAF_CHECK_EQUAL
(
queue
.
pending
.
size
(),
0u
);
}
CAF_TEST
(
ascending
IDs
)
{
CAF_CHECK_EQUAL
(
queue
.
new_id
(),
0u
);
CAF_CHECK_EQUAL
(
queue
.
new_id
(),
1u
);
CAF_CHECK_EQUAL
(
queue
.
new_id
(),
2u
);
CAF_CHECK_EQUAL
(
queue
.
next_undelivered
,
0u
);
}
CAF_TEST
(
push
order
0
-
1
-
2
)
{
acquire_ids
(
3
);
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
push
(
1
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
push
(
2
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
push
order
0
-
2
-
1
)
{
acquire_ids
(
3
);
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
push
(
2
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
1
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
push
order
1
-
0
-
2
)
{
acquire_ids
(
3
);
push
(
1
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
push
(
2
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
push
order
1
-
2
-
0
)
{
acquire_ids
(
3
);
push
(
1
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
2
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
push
order
2
-
0
-
1
)
{
acquire_ids
(
3
);
push
(
2
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
push
(
1
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
push
order
2
-
1
-
0
)
{
acquire_ids
(
3
);
push
(
2
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
1
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
1
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST
(
dropping
)
{
acquire_ids
(
3
);
push
(
2
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
queue
.
drop
(
nullptr
,
1
);
disallow
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
));
push
(
0
);
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
0
));
expect
((
ok_atom
,
int
),
from
(
self
).
to
(
testee
).
with
(
_
,
2
));
}
CAF_TEST_FIXTURE_SCOPE_END
()
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