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
aae2e2b9
Commit
aae2e2b9
authored
Feb 11, 2015
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'topic/ipv6' into develop
parents
20a02150
95a30652
Changes
8
Show whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
323 additions
and
221 deletions
+323
-221
libcaf_io/caf/io/network/acceptor_manager.hpp
libcaf_io/caf/io/network/acceptor_manager.hpp
+0
-3
libcaf_io/caf/io/network/default_multiplexer.hpp
libcaf_io/caf/io/network/default_multiplexer.hpp
+36
-75
libcaf_io/caf/io/network/interfaces.hpp
libcaf_io/caf/io/network/interfaces.hpp
+11
-3
libcaf_io/caf/io/network/manager.hpp
libcaf_io/caf/io/network/manager.hpp
+1
-6
libcaf_io/caf/io/network/stream_manager.hpp
libcaf_io/caf/io/network/stream_manager.hpp
+1
-4
libcaf_io/src/default_multiplexer.cpp
libcaf_io/src/default_multiplexer.cpp
+239
-129
libcaf_io/src/interfaces.cpp
libcaf_io/src/interfaces.cpp
+34
-1
unit_testing/test_unpublish.cpp
unit_testing/test_unpublish.cpp
+1
-0
No files found.
libcaf_io/caf/io/network/acceptor_manager.hpp
View file @
aae2e2b9
...
@@ -31,9 +31,7 @@ namespace network {
...
@@ -31,9 +31,7 @@ namespace network {
* callbacks for incoming connections as well as for error handling.
* callbacks for incoming connections as well as for error handling.
*/
*/
class
acceptor_manager
:
public
manager
{
class
acceptor_manager
:
public
manager
{
public:
public:
~
acceptor_manager
();
~
acceptor_manager
();
/**
/**
...
@@ -41,7 +39,6 @@ class acceptor_manager : public manager {
...
@@ -41,7 +39,6 @@ class acceptor_manager : public manager {
* a new connection is awaiting acceptance.
* a new connection is awaiting acceptance.
*/
*/
virtual
void
new_connection
()
=
0
;
virtual
void
new_connection
()
=
0
;
};
};
}
// namespace network
}
// namespace network
...
...
libcaf_io/caf/io/network/default_multiplexer.hpp
View file @
aae2e2b9
...
@@ -153,12 +153,6 @@ void nonblocking(native_socket fd, bool new_value);
...
@@ -153,12 +153,6 @@ void nonblocking(native_socket fd, bool new_value);
*/
*/
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
();
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
();
/**
* Throws network_error with given error message and
* the platform-specific error code if `add_errno` is `true`.
*/
void
throw_io_failure
(
const
char
*
what
,
bool
add_errno
=
true
);
/**
/**
* Returns true if `fd` is configured as nodelay socket.
* Returns true if `fd` is configured as nodelay socket.
* @throws network_error
* @throws network_error
...
@@ -222,6 +216,11 @@ class event_handler {
...
@@ -222,6 +216,11 @@ class event_handler {
*/
*/
virtual
void
removed_from_loop
(
operation
op
)
=
0
;
virtual
void
removed_from_loop
(
operation
op
)
=
0
;
/**
* Returns the native socket handle for this handler.
*/
virtual
native_socket
fd
()
const
=
0
;
/**
/**
* Returns the `multiplexer` this acceptor belongs to.
* Returns the `multiplexer` this acceptor belongs to.
*/
*/
...
@@ -243,11 +242,6 @@ class event_handler {
...
@@ -243,11 +242,6 @@ class event_handler {
m_eventbf
=
value
;
m_eventbf
=
value
;
}
}
/**
* Returns the native socket handle for this handler.
*/
virtual
native_socket
fd
()
const
=
0
;
protected:
protected:
default_multiplexer
&
m_backend
;
default_multiplexer
&
m_backend
;
int
m_eventbf
;
int
m_eventbf
;
...
@@ -294,12 +288,10 @@ class default_socket {
...
@@ -294,12 +288,10 @@ class default_socket {
using
default_socket_acceptor
=
default_socket
;
using
default_socket_acceptor
=
default_socket
;
class
default_multiplexer
:
public
multiplexer
{
class
default_multiplexer
:
public
multiplexer
{
public:
friend
class
io
::
middleman
;
// disambiguate reference
friend
class
io
::
middleman
;
// disambiguate reference
friend
class
supervisor
;
friend
class
supervisor
;
public:
struct
event
{
struct
event
{
native_socket
fd
;
native_socket
fd
;
int
mask
;
int
mask
;
...
@@ -356,7 +348,6 @@ class default_multiplexer : public multiplexer {
...
@@ -356,7 +348,6 @@ class default_multiplexer : public multiplexer {
void
del
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
);
void
del
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
);
private:
private:
// platform-dependent additional initialization code
// platform-dependent additional initialization code
void
init
();
void
init
();
...
@@ -417,21 +408,18 @@ class default_multiplexer : public multiplexer {
...
@@ -417,21 +408,18 @@ class default_multiplexer : public multiplexer {
std
::
vector
<
event
>
m_events
;
// always sorted by .fd
std
::
vector
<
event
>
m_events
;
// always sorted by .fd
multiplexer_poll_shadow_data
m_shadow
;
multiplexer_poll_shadow_data
m_shadow
;
std
::
pair
<
native_socket
,
native_socket
>
m_pipe
;
std
::
pair
<
native_socket
,
native_socket
>
m_pipe
;
std
::
thread
::
id
m_tid
;
};
};
default_multiplexer
&
get_multiplexer_singleton
();
default_multiplexer
&
get_multiplexer_singleton
();
template
<
class
T
>
template
<
class
T
>
inline
connection_handle
conn_hdl_from_socket
(
const
T
&
sock
)
{
connection_handle
conn_hdl_from_socket
(
const
T
&
sock
)
{
return
connection_handle
::
from_int
(
return
connection_handle
::
from_int
(
int64_from_native_socket
(
sock
.
native_handle
()));
int64_from_native_socket
(
sock
.
native_handle
()));
}
}
template
<
class
T
>
template
<
class
T
>
inline
accept_handle
accept_hdl_from_socket
(
const
T
&
sock
)
{
accept_handle
accept_hdl_from_socket
(
const
T
&
sock
)
{
return
accept_handle
::
from_int
(
return
accept_handle
::
from_int
(
int64_from_native_socket
(
sock
.
native_handle
()));
int64_from_native_socket
(
sock
.
native_handle
()));
}
}
...
@@ -462,17 +450,10 @@ class stream : public event_handler {
...
@@ -462,17 +450,10 @@ class stream : public event_handler {
configure_read
(
receive_policy
::
at_most
(
1024
));
configure_read
(
receive_policy
::
at_most
(
1024
));
}
}
/**
* Returns the `multiplexer` this stream belongs to.
*/
inline
default_multiplexer
&
backend
()
{
return
static_cast
<
default_multiplexer
&>
(
m_sock
.
backend
());
}
/**
/**
* Returns the IO socket.
* Returns the IO socket.
*/
*/
inline
Socket
&
socket_handle
()
{
Socket
&
socket_handle
()
{
return
m_sock
;
return
m_sock
;
}
}
...
@@ -502,8 +483,7 @@ class stream : public event_handler {
...
@@ -502,8 +483,7 @@ class stream : public event_handler {
}
}
/**
/**
* Configures how much data will be provided
* Configures how much data will be provided for the next `consume` callback.
* for the next `consume` callback.
* @warning Must not be called outside the IO multiplexers event loop
* @warning Must not be called outside the IO multiplexers event loop
* once the stream has been started.
* once the stream has been started.
*/
*/
...
@@ -514,7 +494,7 @@ class stream : public event_handler {
...
@@ -514,7 +494,7 @@ class stream : public event_handler {
/**
/**
* Copies data to the write buffer.
* Copies data to the write buffer.
* @
note
Not thread safe.
* @
warning
Not thread safe.
*/
*/
void
write
(
const
void
*
buf
,
size_t
num_bytes
)
{
void
write
(
const
void
*
buf
,
size_t
num_bytes
)
{
CAF_LOG_TRACE
(
"num_bytes: "
<<
num_bytes
);
CAF_LOG_TRACE
(
"num_bytes: "
<<
num_bytes
);
...
@@ -561,7 +541,7 @@ class stream : public event_handler {
...
@@ -561,7 +541,7 @@ class stream : public event_handler {
backend
().
del
(
operation
::
read
,
m_sock
.
fd
(),
this
);
backend
().
del
(
operation
::
read
,
m_sock
.
fd
(),
this
);
}
}
void
handle_event
(
operation
op
)
{
void
handle_event
(
operation
op
)
override
{
CAF_LOG_TRACE
(
"op = "
<<
static_cast
<
int
>
(
op
));
CAF_LOG_TRACE
(
"op = "
<<
static_cast
<
int
>
(
op
));
switch
(
op
)
{
switch
(
op
)
{
case
operation
:
:
read
:
{
case
operation
:
:
read
:
{
...
@@ -610,7 +590,6 @@ class stream : public event_handler {
...
@@ -610,7 +590,6 @@ class stream : public event_handler {
}
}
}
}
protected:
native_socket
fd
()
const
override
{
native_socket
fd
()
const
override
{
return
m_sock
.
fd
();
return
m_sock
.
fd
();
}
}
...
@@ -679,9 +658,7 @@ class stream : public event_handler {
...
@@ -679,9 +658,7 @@ class stream : public event_handler {
*/
*/
template
<
class
SocketAcceptor
>
template
<
class
SocketAcceptor
>
class
acceptor
:
public
event_handler
{
class
acceptor
:
public
event_handler
{
public:
public:
using
socket_type
=
typename
SocketAcceptor
::
socket_type
;
using
socket_type
=
typename
SocketAcceptor
::
socket_type
;
/**
/**
...
@@ -704,7 +681,7 @@ class acceptor : public event_handler {
...
@@ -704,7 +681,7 @@ class acceptor : public event_handler {
/**
/**
* Returns the IO socket.
* Returns the IO socket.
*/
*/
inline
SocketAcceptor
&
socket_handle
()
{
SocketAcceptor
&
socket_handle
()
{
return
m_accept_sock
;
return
m_accept_sock
;
}
}
...
@@ -712,7 +689,7 @@ class acceptor : public event_handler {
...
@@ -712,7 +689,7 @@ class acceptor : public event_handler {
* Returns the accepted socket. This member function should
* Returns the accepted socket. This member function should
* be called only from the `new_connection` callback.
* be called only from the `new_connection` callback.
*/
*/
inline
socket_type
&
accepted_socket
()
{
socket_type
&
accepted_socket
()
{
return
m_sock
;
return
m_sock
;
}
}
...
@@ -763,49 +740,33 @@ class acceptor : public event_handler {
...
@@ -763,49 +740,33 @@ class acceptor : public event_handler {
void
removed_from_loop
(
operation
op
)
override
{
void
removed_from_loop
(
operation
op
)
override
{
CAF_LOG_TRACE
(
"m_accept_sock.fd = "
<<
m_accept_sock
.
fd
()
CAF_LOG_TRACE
(
"m_accept_sock.fd = "
<<
m_accept_sock
.
fd
()
<<
"op = "
<<
static_cast
<
int
>
(
op
));
<<
"op = "
<<
static_cast
<
int
>
(
op
));
if
(
op
==
operation
::
read
)
m_mgr
.
reset
();
if
(
op
==
operation
::
read
)
{
m_mgr
.
reset
();
}
}
}
protected:
native_socket
fd
()
const
override
{
native_socket
fd
()
const
override
{
return
m_accept_sock
.
fd
();
return
m_accept_sock
.
fd
();
}
}
private:
private:
manager_ptr
m_mgr
;
manager_ptr
m_mgr
;
SocketAcceptor
m_accept_sock
;
SocketAcceptor
m_accept_sock
;
socket_type
m_sock
;
socket_type
m_sock
;
};
};
native_socket
new_ipv4_connection_impl
(
const
std
::
string
&
,
uint16_t
);
native_socket
new_tcp_connection_impl
(
const
std
::
string
&
,
uint16_t
,
optional
<
protocol
>
preferred
=
none
);
default_socket
new_ipv4_connection
(
const
std
::
string
&
host
,
uint16_t
port
);
template
<
class
Socket
>
default_socket
new_tcp_connection
(
const
std
::
string
&
host
,
uint16_t
port
);
void
ipv4_connect
(
Socket
&
sock
,
const
std
::
string
&
host
,
uint16_t
port
)
{
sock
=
new_ipv4_connection
(
host
,
port
);
}
std
::
pair
<
native_socket
,
uint16_t
>
std
::
pair
<
native_socket
,
uint16_t
>
new_
ipv4
_acceptor_impl
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse_addr
);
new_
tcp
_acceptor_impl
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse_addr
);
std
::
pair
<
default_socket_acceptor
,
uint16_t
>
std
::
pair
<
default_socket_acceptor
,
uint16_t
>
new_
ipv4
_acceptor
(
uint16_t
port
,
const
char
*
addr
=
nullptr
,
new_
tcp
_acceptor
(
uint16_t
port
,
const
char
*
addr
=
nullptr
,
bool
reuse_addr
=
false
);
bool
reuse_addr
=
false
);
template
<
class
SocketAcceptor
>
uint16_t
ipv4_bind
(
SocketAcceptor
&
sock
,
uint16_t
port
,
const
char
*
addr
=
nullptr
)
{
CAF_LOGF_TRACE
(
CAF_ARG
(
port
));
auto
acceptor
=
new_ipv4_acceptor
(
port
,
addr
);
sock
=
std
::
move
(
acceptor
.
first
);
return
acceptor
.
second
;
}
}
// namespace network
}
// namespace network
}
// namespace io
}
// namespace io
}
// namespace caf
}
// namespace caf
...
...
libcaf_io/caf/io/network/interfaces.hpp
View file @
aae2e2b9
...
@@ -17,14 +17,16 @@
...
@@ -17,14 +17,16 @@
* http://www.boost.org/LICENSE_1_0.txt. *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
******************************************************************************/
#ifndef CAF_IO_NETWORK_
ADDRESS
ES_HPP
#ifndef CAF_IO_NETWORK_
INTERFAC
ES_HPP
#define CAF_IO_NETWORK_
ADDRESS
ES_HPP
#define CAF_IO_NETWORK_
INTERFAC
ES_HPP
#include <map>
#include <map>
#include <vector>
#include <vector>
#include <string>
#include <string>
#include <utility>
#include <utility>
#include "caf/optional.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/protocol.hpp"
namespace
caf
{
namespace
caf
{
...
@@ -57,10 +59,16 @@ class interfaces {
...
@@ -57,10 +59,16 @@ class interfaces {
*/
*/
static
std
::
vector
<
std
::
string
>
list_addresses
(
protocol
proc
,
static
std
::
vector
<
std
::
string
>
list_addresses
(
protocol
proc
,
bool
include_localhost
=
true
);
bool
include_localhost
=
true
);
/**
* Returns a native IPv4 or IPv6 translation of `host`.
**/
static
optional
<
std
::
pair
<
std
::
string
,
protocol
>>
native_address
(
const
std
::
string
&
host
,
optional
<
protocol
>
preferred
=
none
);
};
};
}
// namespace network
}
// namespace network
}
// namespace io
}
// namespace io
}
// namespace caf
}
// namespace caf
#endif // CAF_IO_NETWORK_
ADDRESS
ES_HPP
#endif // CAF_IO_NETWORK_
INTERFAC
ES_HPP
libcaf_io/caf/io/network/manager.hpp
View file @
aae2e2b9
...
@@ -35,7 +35,7 @@ namespace network {
...
@@ -35,7 +35,7 @@ namespace network {
*/
*/
class
manager
:
public
ref_counted
{
class
manager
:
public
ref_counted
{
public:
public:
virtual
~
manager
();
~
manager
();
/**
/**
* Causes the manager to stop read operations on its IO device.
* Causes the manager to stop read operations on its IO device.
...
@@ -49,11 +49,6 @@ class manager : public ref_counted {
...
@@ -49,11 +49,6 @@ class manager : public ref_counted {
virtual
void
io_failure
(
operation
op
)
=
0
;
virtual
void
io_failure
(
operation
op
)
=
0
;
};
};
/**
* @relates manager
*/
using
manager_ptr
=
intrusive_ptr
<
manager
>
;
}
// namespace network
}
// namespace network
}
// namespace io
}
// namespace io
}
// namespace caf
}
// namespace caf
...
...
libcaf_io/caf/io/network/stream_manager.hpp
View file @
aae2e2b9
...
@@ -33,16 +33,13 @@ namespace network {
...
@@ -33,16 +33,13 @@ namespace network {
* for incoming data as well as for error handling.
* for incoming data as well as for error handling.
*/
*/
class
stream_manager
:
public
manager
{
class
stream_manager
:
public
manager
{
public:
public:
~
stream_manager
();
virtual
~
stream_manager
();
/**
/**
* Called by the underlying IO device whenever it received data.
* Called by the underlying IO device whenever it received data.
*/
*/
virtual
void
consume
(
const
void
*
data
,
size_t
num_bytes
)
=
0
;
virtual
void
consume
(
const
void
*
data
,
size_t
num_bytes
)
=
0
;
};
};
}
// namespace network
}
// namespace network
...
...
libcaf_io/src/default_multiplexer.cpp
View file @
aae2e2b9
...
@@ -20,11 +20,15 @@
...
@@ -20,11 +20,15 @@
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/optional.hpp"
#include "caf/exception.hpp"
#include "caf/exception.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/interfaces.hpp"
#ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <winsock2.h>
# include <ws2tcpip.h>
/* socklen_t, et al (MSVC20xx) */
# include <ws2tcpip.h>
/* socklen_t, et al (MSVC20xx) */
...
@@ -44,6 +48,7 @@
...
@@ -44,6 +48,7 @@
using
std
::
string
;
using
std
::
string
;
namespace
{
namespace
{
#ifdef CAF_MACOS
#ifdef CAF_MACOS
constexpr
int
no_sigpipe_flag
=
SO_NOSIGPIPE
;
constexpr
int
no_sigpipe_flag
=
SO_NOSIGPIPE
;
#elif defined(CAF_WINDOWS)
#elif defined(CAF_WINDOWS)
...
@@ -51,6 +56,46 @@ namespace {
...
@@ -51,6 +56,46 @@ namespace {
#else // BSD or Linux
#else // BSD or Linux
constexpr
int
no_sigpipe_flag
=
MSG_NOSIGNAL
;
constexpr
int
no_sigpipe_flag
=
MSG_NOSIGNAL
;
#endif
#endif
// safe ourselves some typing
constexpr
auto
ipv4
=
caf
::
io
::
network
::
protocol
::
ipv4
;
constexpr
auto
ipv6
=
caf
::
io
::
network
::
protocol
::
ipv6
;
// predicate for `ccall` meaning "expected result of f is 0"
bool
cc_zero
(
int
value
)
{
return
value
==
0
;
}
// predicate for `ccall` meaning "expected result of f is 1"
bool
cc_one
(
int
value
)
{
return
value
==
1
;
}
// predicate for `ccall` meaning "expected result of f is not -1"
bool
cc_not_minus1
(
int
value
)
{
return
value
!=
-
1
;
}
// predicate for `ccall` meaning "expected result of f is a valid socket"
bool
cc_valid_socket
(
caf
::
io
::
network
::
native_socket
fd
)
{
return
fd
!=
caf
::
io
::
network
::
invalid_native_socket
;
}
// calls a C function and throws a `network_error` if `p` returns false
template
<
class
Predicate
,
class
F
,
class
...
Ts
>
auto
ccall
(
Predicate
p
,
const
char
*
errmsg
,
F
f
,
Ts
&&
...
args
)
->
decltype
(
f
(
std
::
forward
<
Ts
>
(
args
)...))
{
using
namespace
caf
::
io
::
network
;
auto
result
=
f
(
std
::
forward
<
Ts
>
(
args
)...);
if
(
!
p
(
result
))
{
std
::
ostringstream
oss
;
oss
<<
errmsg
<<
": "
<<
last_socket_error_as_string
()
<<
" [errno: "
<<
last_socket_error
()
<<
"]"
;
throw
caf
::
network_error
(
oss
.
str
());
}
return
result
;
}
}
// namespace <anonymous>
}
// namespace <anonymous>
namespace
caf
{
namespace
caf
{
...
@@ -69,15 +114,10 @@ namespace network {
...
@@ -69,15 +114,10 @@ namespace network {
void
nonblocking
(
native_socket
fd
,
bool
new_value
)
{
void
nonblocking
(
native_socket
fd
,
bool
new_value
)
{
// read flags for fd
// read flags for fd
auto
rf
=
fcntl
(
fd
,
F_GETFL
,
0
);
auto
rf
=
ccall
(
cc_not_minus1
,
"cannot read flags"
,
fcntl
,
fd
,
F_GETFL
,
0
);
if
(
rf
==
-
1
)
{
throw_io_failure
(
"unable to read socket flags"
);
}
// calculate and set new flags
// calculate and set new flags
auto
wf
=
new_value
?
(
rf
|
O_NONBLOCK
)
:
(
rf
&
(
~
(
O_NONBLOCK
)));
auto
wf
=
new_value
?
(
rf
|
O_NONBLOCK
)
:
(
rf
&
(
~
(
O_NONBLOCK
)));
if
(
fcntl
(
fd
,
F_SETFL
,
wf
)
<
0
)
{
ccall
(
cc_not_minus1
,
"cannot set flags"
,
fcntl
,
fd
,
F_SETFL
,
wf
);
throw_io_failure
(
"unable to set file descriptor flags"
);
}
}
}
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
()
{
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
()
{
...
@@ -117,17 +157,7 @@ namespace network {
...
@@ -117,17 +157,7 @@ namespace network {
void
nonblocking
(
native_socket
fd
,
bool
new_value
)
{
void
nonblocking
(
native_socket
fd
,
bool
new_value
)
{
u_long
mode
=
new_value
?
1
:
0
;
u_long
mode
=
new_value
?
1
:
0
;
if
(
ioctlsocket
(
fd
,
FIONBIO
,
&
mode
)
<
0
)
{
ccall
(
cc_zero
,
"unable to set FIONBIO"
,
ioctlsocket
,
fd
,
FIONBIO
,
&
mode
);
throw_io_failure
(
"unable to set FIONBIO"
);
}
}
// calls a C function and converts its returned error code to an exception
template
<
class
F
,
class
...
Ts
>
inline
void
ccall
(
const
char
*
errmsg
,
F
f
,
Ts
&&
...
args
)
{
if
(
f
(
std
::
forward
<
Ts
>
(
args
)...)
!=
0
)
{
throw_io_failure
(
errmsg
);
}
}
}
/**************************************************************************\
/**************************************************************************\
...
@@ -164,10 +194,8 @@ namespace network {
...
@@ -164,10 +194,8 @@ namespace network {
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
()
{
std
::
pair
<
native_socket
,
native_socket
>
create_pipe
()
{
socklen_t
addrlen
=
sizeof
(
sockaddr_in
);
socklen_t
addrlen
=
sizeof
(
sockaddr_in
);
native_socket
socks
[
2
]
=
{
invalid_native_socket
,
invalid_native_socket
};
native_socket
socks
[
2
]
=
{
invalid_native_socket
,
invalid_native_socket
};
auto
listener
=
socket
(
AF_INET
,
SOCK_STREAM
,
IPPROTO_TCP
);
auto
listener
=
ccall
(
cc_valid_socket
,
"socket() failed"
,
socket
,
AF_INET
,
if
(
listener
==
invalid_native_socket
)
{
SOCK_STREAM
,
IPPROTO_TCP
);
throw_io_failure
(
"socket() failed"
);
}
union
{
union
{
sockaddr_in
inaddr
;
sockaddr_in
inaddr
;
sockaddr
addr
;
sockaddr
addr
;
...
@@ -190,30 +218,29 @@ namespace network {
...
@@ -190,30 +218,29 @@ namespace network {
});
});
// bind listener to a local port
// bind listener to a local port
int
reuse
=
1
;
int
reuse
=
1
;
ccall
(
"setsockopt() failed"
,
::
setsockopt
,
listener
,
SOL_SOCKET
,
ccall
(
cc_zero
,
"setsockopt() failed"
,
setsockopt
,
listener
,
SOL_SOCKET
,
SO_REUSEADDR
,
reinterpret_cast
<
char
*>
(
&
reuse
),
SO_REUSEADDR
,
reinterpret_cast
<
char
*>
(
&
reuse
),
static_cast
<
socklen_t
>
(
sizeof
(
reuse
)));
int
{
sizeof
(
reuse
)});
ccall
(
"bind() failed"
,
::
bind
,
listener
,
&
a
.
addr
,
sizeof
(
a
.
inaddr
));
ccall
(
cc_zero
,
"bind() failed"
,
bind
,
listener
,
&
a
.
addr
,
int
{
sizeof
(
a
.
inaddr
)});
// read the port in use: win32 getsockname may only set the port number
// read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx):
// (http://msdn.microsoft.com/library/ms738543.aspx):
memset
(
&
a
,
0
,
sizeof
(
a
));
memset
(
&
a
,
0
,
sizeof
(
a
));
ccall
(
"getsockname() failed"
,
::
getsockname
,
listener
,
&
a
.
addr
,
&
addrlen
);
ccall
(
cc_zero
,
"getsockname() failed"
,
getsockname
,
listener
,
&
a
.
addr
,
&
addrlen
);
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_family
=
AF_INET
;
a
.
inaddr
.
sin_family
=
AF_INET
;
// set listener to listen mode
// set listener to listen mode
ccall
(
"listen() failed"
,
::
listen
,
listener
,
1
);
ccall
(
cc_zero
,
"listen() failed"
,
listen
,
listener
,
1
);
// create read-only end of the pipe
// create read-only end of the pipe
DWORD
flags
=
0
;
DWORD
flags
=
0
;
auto
read_fd
=
WSASocket
(
AF_INET
,
SOCK_STREAM
,
0
,
NULL
,
0
,
flags
);
auto
read_fd
=
ccall
(
cc_valid_socket
,
WSASocket
,
AF_INET
,
SOCK_STREAM
,
if
(
read_fd
==
invalid_native_socket
)
{
0
,
NULL
,
0
,
flags
);
throw_io_failure
(
"cannot create read handle: WSASocket() failed"
);
ccall
(
cc_zero
,
"connect() failed"
,
connect
,
read_fd
,
}
&
a
.
addr
,
int
{
sizeof
(
a
.
inaddr
)});
ccall
(
"connect() failed"
,
::
connect
,
read_fd
,
&
a
.
addr
,
sizeof
(
a
.
inaddr
));
// get write-only end of the pipe
// get write-only end of the pipe
auto
write_fd
=
accept
(
listener
,
NULL
,
NULL
);
auto
write_fd
=
ccall
(
cc_valid_socket
,
"accept() failed"
,
if
(
write_fd
==
invalid_native_socket
)
{
accept
,
listener
,
NULL
,
NULL
);
throw_io_failure
(
"cannot create write handle: accept() failed"
);
}
closesocket
(
listener
);
closesocket
(
listener
);
success
=
true
;
success
=
true
;
return
{
read_fd
,
write_fd
};
return
{
read_fd
,
write_fd
};
...
@@ -535,7 +562,8 @@ int del_flag(operation op, int bf) {
...
@@ -535,7 +562,8 @@ int del_flag(operation op, int bf) {
return
0
;
return
0
;
}
}
void
default_multiplexer
::
add
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
)
{
void
default_multiplexer
::
add
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
)
{
CAF_REQUIRE
(
fd
!=
invalid_native_socket
);
CAF_REQUIRE
(
fd
!=
invalid_native_socket
);
// ptr == nullptr is only allowed to store our pipe read handle
// ptr == nullptr is only allowed to store our pipe read handle
// and the pipe read handle is added in the ctor (not allowed here)
// and the pipe read handle is added in the ctor (not allowed here)
...
@@ -545,7 +573,8 @@ void default_multiplexer::add(operation op, native_socket fd, event_handler* ptr
...
@@ -545,7 +573,8 @@ void default_multiplexer::add(operation op, native_socket fd, event_handler* ptr
new_event
(
add_flag
,
op
,
fd
,
ptr
);
new_event
(
add_flag
,
op
,
fd
,
ptr
);
}
}
void
default_multiplexer
::
del
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
)
{
void
default_multiplexer
::
del
(
operation
op
,
native_socket
fd
,
event_handler
*
ptr
)
{
CAF_REQUIRE
(
fd
!=
invalid_native_socket
);
CAF_REQUIRE
(
fd
!=
invalid_native_socket
);
// ptr == nullptr is only allowed when removing our pipe read handle
// ptr == nullptr is only allowed when removing our pipe read handle
CAF_REQUIRE
(
ptr
!=
nullptr
||
fd
==
m_pipe
.
first
);
CAF_REQUIRE
(
ptr
!=
nullptr
||
fd
==
m_pipe
.
first
);
...
@@ -672,14 +701,14 @@ void default_multiplexer::init() {
...
@@ -672,14 +701,14 @@ void default_multiplexer::init() {
}
}
default_multiplexer
::~
default_multiplexer
()
{
default_multiplexer
::~
default_multiplexer
()
{
# ifdef CAF_WINDOWS
WSACleanup
();
# endif
if
(
m_epollfd
!=
invalid_native_socket
)
{
if
(
m_epollfd
!=
invalid_native_socket
)
{
closesocket
(
m_epollfd
);
closesocket
(
m_epollfd
);
}
}
closesocket
(
m_pipe
.
first
);
closesocket
(
m_pipe
.
first
);
closesocket
(
m_pipe
.
second
);
closesocket
(
m_pipe
.
second
);
# ifdef CAF_WINDOWS
WSACleanup
();
# endif
}
}
void
default_multiplexer
::
dispatch_runnable
(
runnable_ptr
ptr
)
{
void
default_multiplexer
::
dispatch_runnable
(
runnable_ptr
ptr
)
{
...
@@ -732,7 +761,8 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
...
@@ -732,7 +761,8 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
return
ptr
->
hdl
();
return
ptr
->
hdl
();
}
}
accept_handle
default_multiplexer
::
add_tcp_doorman
(
broker
*
self
,
accept_handle
default_multiplexer
::
add_tcp_doorman
(
broker
*
self
,
default_socket_acceptor
&&
sock
)
{
default_socket_acceptor
&&
sock
)
{
CAF_LOG_TRACE
(
"sock.fd = "
<<
sock
.
fd
());
CAF_LOG_TRACE
(
"sock.fd = "
<<
sock
.
fd
());
CAF_REQUIRE
(
sock
.
fd
()
!=
network
::
invalid_native_socket
);
CAF_REQUIRE
(
sock
.
fd
()
!=
network
::
invalid_native_socket
);
...
@@ -745,10 +775,9 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
...
@@ -745,10 +775,9 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
}
}
void
new_connection
()
override
{
void
new_connection
()
override
{
auto
&
dm
=
m_acceptor
.
backend
();
auto
&
dm
=
m_acceptor
.
backend
();
accept_msg
().
handle
=
dm
.
add_tcp_scribe
(
parent
(),
accept_msg
().
handle
std
::
move
(
m_acceptor
.
accepted_socket
()));
=
dm
.
add_tcp_scribe
(
parent
(),
std
::
move
(
m_acceptor
.
accepted_socket
()));
parent
()
->
invoke_message
(
invalid_actor_addr
,
parent
()
->
invoke_message
(
invalid_actor_addr
,
invalid_message_id
,
invalid_message_id
,
m_accept_msg
);
m_accept_msg
);
}
}
void
stop_reading
()
override
{
void
stop_reading
()
override
{
...
@@ -768,7 +797,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
...
@@ -768,7 +797,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
connection_handle
default_multiplexer
::
new_tcp_scribe
(
const
std
::
string
&
host
,
connection_handle
default_multiplexer
::
new_tcp_scribe
(
const
std
::
string
&
host
,
uint16_t
port
)
{
uint16_t
port
)
{
auto
fd
=
new_
ipv4
_connection_impl
(
host
,
port
);
auto
fd
=
new_
tcp
_connection_impl
(
host
,
port
);
return
connection_handle
::
from_int
(
int64_from_native_socket
(
fd
));
return
connection_handle
::
from_int
(
int64_from_native_socket
(
fd
));
}
}
...
@@ -785,14 +814,13 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
...
@@ -785,14 +814,13 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
connection_handle
default_multiplexer
::
add_tcp_scribe
(
broker
*
self
,
connection_handle
default_multiplexer
::
add_tcp_scribe
(
broker
*
self
,
const
std
::
string
&
host
,
const
std
::
string
&
host
,
uint16_t
port
)
{
uint16_t
port
)
{
return
add_tcp_scribe
(
self
,
new_
ipv4
_connection
(
host
,
port
));
return
add_tcp_scribe
(
self
,
new_
tcp
_connection
(
host
,
port
));
}
}
std
::
pair
<
accept_handle
,
uint16_t
>
std
::
pair
<
accept_handle
,
uint16_t
>
default_multiplexer
::
new_tcp_doorman
(
uint16_t
port
,
const
char
*
in
,
default_multiplexer
::
new_tcp_doorman
(
uint16_t
port
,
const
char
*
in
,
bool
reuse_addr
)
{
bool
reuse_addr
)
{
auto
res
=
new_
ipv4
_acceptor_impl
(
port
,
in
,
reuse_addr
);
auto
res
=
new_
tcp
_acceptor_impl
(
port
,
in
,
reuse_addr
);
return
{
accept_handle
::
from_int
(
int64_from_native_socket
(
res
.
first
)),
return
{
accept_handle
::
from_int
(
int64_from_native_socket
(
res
.
first
)),
res
.
second
};
res
.
second
};
}
}
...
@@ -809,7 +837,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
...
@@ -809,7 +837,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
std
::
pair
<
accept_handle
,
uint16_t
>
std
::
pair
<
accept_handle
,
uint16_t
>
default_multiplexer
::
add_tcp_doorman
(
broker
*
self
,
uint16_t
port
,
default_multiplexer
::
add_tcp_doorman
(
broker
*
self
,
uint16_t
port
,
const
char
*
host
,
bool
reuse_addr
)
{
const
char
*
host
,
bool
reuse_addr
)
{
auto
acceptor
=
new_
ipv4
_acceptor
(
port
,
host
,
reuse_addr
);
auto
acceptor
=
new_
tcp
_acceptor
(
port
,
host
,
reuse_addr
);
auto
bound_port
=
acceptor
.
second
;
auto
bound_port
=
acceptor
.
second
;
return
{
add_tcp_doorman
(
self
,
std
::
move
(
acceptor
.
first
)),
bound_port
};
return
{
add_tcp_doorman
(
self
,
std
::
move
(
acceptor
.
first
)),
bound_port
};
}
}
...
@@ -818,22 +846,11 @@ default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
...
@@ -818,22 +846,11 @@ default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
* platform-independent implementations (finally) *
* platform-independent implementations (finally) *
******************************************************************************/
******************************************************************************/
void
throw_io_failure
(
const
char
*
what
,
bool
add_errno
)
{
if
(
add_errno
)
{
std
::
ostringstream
oss
;
oss
<<
what
<<
": "
<<
last_socket_error_as_string
()
<<
" [errno: "
<<
last_socket_error
()
<<
"]"
;
throw
network_error
(
oss
.
str
());
}
throw
network_error
(
what
);
}
void
tcp_nodelay
(
native_socket
fd
,
bool
new_value
)
{
void
tcp_nodelay
(
native_socket
fd
,
bool
new_value
)
{
int
flag
=
new_value
?
1
:
0
;
int
flag
=
new_value
?
1
:
0
;
if
(
setsockopt
(
fd
,
IPPROTO_TCP
,
TCP_NODELAY
,
ccall
(
cc_zero
,
"unable to set TCP_NODELAY"
,
setsockopt
,
fd
,
IPPROTO_TCP
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
flag
),
sizeof
(
flag
))
<
0
)
{
TCP_NODELAY
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
flag
),
throw_io_failure
(
"unable to set TCP_NODELAY"
);
socklen_t
{
sizeof
(
flag
)});
}
}
}
bool
is_error
(
ssize_t
res
,
bool
is_nonblock
)
{
bool
is_error
(
ssize_t
res
,
bool
is_nonblock
)
{
...
@@ -912,7 +929,8 @@ default_socket::default_socket(default_multiplexer& ref, native_socket sockfd)
...
@@ -912,7 +929,8 @@ default_socket::default_socket(default_multiplexer& ref, native_socket sockfd)
}
}
default_socket
::
default_socket
(
default_socket
&&
other
)
default_socket
::
default_socket
(
default_socket
&&
other
)
:
m_parent
(
other
.
m_parent
),
m_fd
(
other
.
m_fd
)
{
:
m_parent
(
other
.
m_parent
),
m_fd
(
other
.
m_fd
)
{
other
.
m_fd
=
invalid_native_socket
;
other
.
m_fd
=
invalid_native_socket
;
}
}
...
@@ -934,115 +952,207 @@ void default_socket::close_read() {
...
@@ -934,115 +952,207 @@ void default_socket::close_read() {
}
}
}
}
struct
socket_guard
{
class
socket_guard
{
public:
public:
socket_guard
(
native_socket
fd
)
:
m_fd
(
fd
)
{
socket_guard
(
native_socket
fd
)
:
m_fd
(
fd
)
{
// nop
// nop
}
}
~
socket_guard
()
{
~
socket_guard
()
{
if
(
m_fd
!=
invalid_native_socket
)
if
(
m_fd
!=
invalid_native_socket
)
closesocket
(
m_fd
);
closesocket
(
m_fd
);
}
}
void
release
()
{
native_socket
release
()
{
auto
fd
=
m_fd
;
m_fd
=
invalid_native_socket
;
m_fd
=
invalid_native_socket
;
return
fd
;
}
}
private:
private:
native_socket
m_fd
;
native_socket
m_fd
;
};
};
native_socket
new_ipv4_connection_impl
(
const
std
::
string
&
host
,
uint16_t
port
)
{
in_addr
&
addr_of
(
sockaddr_in
&
what
)
{
return
what
.
sin_addr
;
}
sa_family_t
&
family_of
(
sockaddr_in
&
what
)
{
return
what
.
sin_family
;
}
in_port_t
&
port_of
(
sockaddr_in
&
what
)
{
return
what
.
sin_port
;
}
in6_addr
&
addr_of
(
sockaddr_in6
&
what
)
{
return
what
.
sin6_addr
;
}
sa_family_t
&
family_of
(
sockaddr_in6
&
what
)
{
return
what
.
sin6_family
;
}
in_port_t
&
port_of
(
sockaddr_in6
&
what
)
{
return
what
.
sin6_port
;
}
in_port_t
&
port_of
(
sockaddr
&
what
)
{
switch
(
what
.
sa_family
)
{
case
AF_INET
:
return
port_of
(
reinterpret_cast
<
sockaddr_in
&>
(
what
));
case
AF_INET6
:
return
port_of
(
reinterpret_cast
<
sockaddr_in6
&>
(
what
));
default:
break
;
}
throw
std
::
invalid_argument
(
"invalid protocol family"
);
}
template
<
int
Family
>
bool
ip_connect
(
native_socket
fd
,
const
std
::
string
&
host
,
uint16_t
port
)
{
static_assert
(
Family
==
AF_INET
||
Family
==
AF_INET6
,
"invalid family"
);
using
sockaddr_type
=
typename
std
::
conditional
<
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
sockaddr_type
sa
;
memset
(
&
sa
,
0
,
sizeof
(
sockaddr_type
));
inet_pton
(
Family
,
host
.
c_str
(),
&
addr_of
(
sa
));
family_of
(
sa
)
=
Family
;
port_of
(
sa
)
=
htons
(
port
);
return
connect
(
fd
,
reinterpret_cast
<
const
sockaddr
*>
(
&
sa
),
sizeof
(
sa
))
==
0
;
}
native_socket
new_tcp_connection_impl
(
const
std
::
string
&
host
,
uint16_t
port
,
optional
<
protocol
>
preferred
)
{
CAF_LOGF_TRACE
(
CAF_ARG
(
host
)
<<
", "
<<
CAF_ARG
(
port
));
CAF_LOGF_TRACE
(
CAF_ARG
(
host
)
<<
", "
<<
CAF_ARG
(
port
));
CAF_LOGF_INFO
(
"try to connect to "
<<
host
<<
" on port "
<<
port
);
CAF_LOGF_INFO
(
"try to connect to "
<<
host
<<
" on port "
<<
port
);
# ifdef CAF_WINDOWS
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
// make sure TCP has been initialized via WSAStartup
get_multiplexer_singleton
();
get_multiplexer_singleton
();
# endif
# endif
sockaddr_in
serv_addr
;
auto
res
=
interfaces
::
native_address
(
host
,
preferred
);
hostent
*
server
;
if
(
!
res
)
{
native_socket
fd
=
socket
(
AF_INET
,
SOCK_STREAM
,
0
);
throw
network_error
(
"no such host: "
+
host
);
if
(
fd
==
invalid_native_socket
)
{
}
throw
network_error
(
"socket creation failed"
);
auto
proto
=
res
->
second
;
}
CAF_REQUIRE
(
proto
==
ipv4
||
proto
==
ipv6
);
auto
fd
=
ccall
(
cc_valid_socket
,
"socket creation failed"
,
socket
,
proto
==
ipv4
?
AF_INET
:
AF_INET6
,
SOCK_STREAM
,
0
);
socket_guard
sguard
(
fd
);
socket_guard
sguard
(
fd
);
server
=
gethostbyname
(
host
.
c_str
());
if
(
proto
==
ipv6
)
{
if
(
!
server
)
{
if
(
ip_connect
<
AF_INET6
>
(
fd
,
res
->
first
,
port
))
{
std
::
string
errstr
=
"no such host: "
;
return
fd
;
errstr
+=
host
;
}
throw
network_error
(
std
::
move
(
errstr
));
// IPv4 fallback
}
return
new_tcp_connection_impl
(
host
,
port
,
ipv4
);
memset
(
&
serv_addr
,
0
,
sizeof
(
serv_addr
));
}
serv_addr
.
sin_family
=
AF_INET
;
if
(
!
ip_connect
<
AF_INET
>
(
fd
,
res
->
first
,
port
))
{
memmove
(
&
serv_addr
.
sin_addr
.
s_addr
,
server
->
h_addr
,
static_cast
<
size_t
>
(
server
->
h_length
));
serv_addr
.
sin_port
=
htons
(
port
);
CAF_LOGF_DEBUG
(
"call connect()"
);
if
(
connect
(
fd
,
reinterpret_cast
<
const
sockaddr
*>
(
&
serv_addr
),
sizeof
(
serv_addr
))
!=
0
)
{
CAF_LOGF_ERROR
(
"could not connect to to "
<<
host
<<
" on port "
<<
port
);
CAF_LOGF_ERROR
(
"could not connect to to "
<<
host
<<
" on port "
<<
port
);
throw
network_error
(
"could not connect to
host"
);
throw
network_error
(
"could not connect to
"
+
host
);
}
}
sguard
.
release
();
return
sguard
.
release
();
return
fd
;
}
}
default_socket
new_
ipv4
_connection
(
const
std
::
string
&
host
,
uint16_t
port
)
{
default_socket
new_
tcp
_connection
(
const
std
::
string
&
host
,
uint16_t
port
)
{
auto
&
backend
=
get_multiplexer_singleton
();
auto
&
backend
=
get_multiplexer_singleton
();
return
default_socket
{
backend
,
new_
ipv4
_connection_impl
(
host
,
port
)};
return
default_socket
{
backend
,
new_
tcp
_connection_impl
(
host
,
port
)};
}
}
std
::
pair
<
native_socket
,
uint16_t
>
template
<
class
SockAddrType
>
new_ipv4_acceptor_impl
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse_addr
)
{
void
read_port
(
native_socket
fd
,
SockAddrType
&
sa
)
{
socklen_t
len
=
sizeof
(
SockAddrType
);
ccall
(
cc_zero
,
"read_port failed"
,
getsockname
,
fd
,
reinterpret_cast
<
sockaddr
*>
(
&
sa
),
&
len
);
}
void
set_inaddr_any
(
native_socket
,
sockaddr_in
&
sa
)
{
sa
.
sin_addr
.
s_addr
=
INADDR_ANY
;
}
void
set_inaddr_any
(
native_socket
fd
,
sockaddr_in6
&
sa
)
{
sa
.
sin6_addr
=
in6addr_any
;
// also accept ipv4 requests on this socket
int
off
=
0
;
ccall
(
cc_zero
,
"unable to unset IPV6_V6ONLY"
,
setsockopt
,
fd
,
IPPROTO_IPV6
,
IPV6_V6ONLY
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
off
),
socklen_t
{
sizeof
(
off
)});
}
template
<
int
Family
>
uint16_t
new_ip_acceptor_impl
(
native_socket
fd
,
uint16_t
port
,
const
char
*
addr
)
{
static_assert
(
Family
==
AF_INET
||
Family
==
AF_INET6
,
"invalid family"
);
CAF_LOGF_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
CAF_LOGF_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
# ifdef CAF_WINDOWS
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
// make sure TCP has been initialized via WSAStartup
get_multiplexer_singleton
();
get_multiplexer_singleton
();
# endif
# endif
native_socket
fd
=
socket
(
AF_INET
,
SOCK_STREAM
,
0
);
using
sockaddr_type
=
if
(
fd
==
invalid_native_socket
)
{
typename
std
::
conditional
<
throw
network_error
(
"could not create server socket"
);
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
sockaddr_type
sa
;
memset
(
&
sa
,
0
,
sizeof
(
sockaddr_type
));
family_of
(
sa
)
=
Family
;
if
(
!
addr
)
{
set_inaddr_any
(
fd
,
sa
);
}
else
{
ccall
(
cc_one
,
"invalid IP address"
,
inet_pton
,
Family
,
addr
,
&
addr_of
(
sa
));
}
}
port_of
(
sa
)
=
htons
(
port
);
ccall
(
cc_zero
,
"cannot bind socket"
,
bind
,
fd
,
reinterpret_cast
<
sockaddr
*>
(
&
sa
),
socklen_t
{
sizeof
(
sa
)});
read_port
(
fd
,
sa
);
return
ntohs
(
port_of
(
sa
));
}
std
::
pair
<
native_socket
,
uint16_t
>
new_tcp_acceptor_impl
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse_addr
)
{
CAF_LOGF_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
get_multiplexer_singleton
();
# endif
protocol
proto
=
ipv6
;
if
(
addr
)
{
auto
addrs
=
interfaces
::
native_address
(
addr
);
if
(
!
addrs
)
{
std
::
string
errmsg
=
"invalid IP address: "
;
errmsg
+=
addr
;
throw
network_error
(
errmsg
);
}
proto
=
addrs
->
second
;
CAF_REQUIRE
(
proto
==
ipv4
||
proto
==
ipv6
);
}
auto
fd
=
ccall
(
cc_valid_socket
,
"could not create server socket"
,
socket
,
proto
==
ipv4
?
AF_INET
:
AF_INET6
,
SOCK_STREAM
,
0
);
// sguard closes the socket in case of exception
// sguard closes the socket in case of exception
socket_guard
sguard
(
fd
);
socket_guard
sguard
(
fd
);
if
(
reuse_addr
)
{
if
(
reuse_addr
)
{
int
on
=
1
;
int
on
=
1
;
if
(
setsockopt
(
fd
,
SOL_SOCKET
,
SO_REUSEADDR
,
ccall
(
cc_zero
,
"unable to set SO_REUSEADDR"
,
setsockopt
,
fd
,
SOL_SOCKET
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
on
),
sizeof
(
on
))
<
0
)
{
SO_REUSEADDR
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
on
),
throw_io_failure
(
"unable to set SO_REUSEADDR"
);
socklen_t
{
sizeof
(
on
)});
}
}
struct
sockaddr_in
serv_addr
;
memset
(
&
serv_addr
,
0
,
sizeof
(
serv_addr
));
serv_addr
.
sin_family
=
AF_INET
;
if
(
!
addr
)
{
serv_addr
.
sin_addr
.
s_addr
=
INADDR_ANY
;
}
else
if
(
::
inet_pton
(
AF_INET
,
addr
,
&
serv_addr
.
sin_addr
)
<=
0
)
{
throw
network_error
(
"invalid IPv4 address"
);
}
serv_addr
.
sin_port
=
htons
(
port
);
if
(
bind
(
fd
,
reinterpret_cast
<
sockaddr
*>
(
&
serv_addr
),
sizeof
(
serv_addr
))
<
0
)
{
throw
bind_failure
(
last_socket_error_as_string
());
}
if
(
listen
(
fd
,
SOMAXCONN
)
!=
0
)
{
throw
network_error
(
"listen() failed: "
+
last_socket_error_as_string
());
}
if
(
port
==
0
)
{
socklen_t
len
=
sizeof
(
serv_addr
);
if
(
getsockname
(
fd
,
reinterpret_cast
<
sockaddr
*>
(
&
serv_addr
),
&
len
)
<
0
)
{
throw
network_error
(
"getsockname(): "
+
last_socket_error_as_string
());
}
}
}
auto
p
=
proto
==
ipv4
?
new_ip_acceptor_impl
<
AF_INET
>
(
fd
,
port
,
addr
)
:
new_ip_acceptor_impl
<
AF_INET6
>
(
fd
,
port
,
addr
);
ccall
(
cc_zero
,
"listen() failed"
,
listen
,
fd
,
SOMAXCONN
);
// ok, no exceptions so far
// ok, no exceptions so far
sguard
.
release
();
CAF_LOGF_DEBUG
(
"sockfd = "
<<
fd
<<
", port = "
<<
p
);
CAF_LOGF_DEBUG
(
"sockfd = "
<<
fd
<<
", port = "
<<
ntohs
(
serv_addr
.
sin_port
));
return
{
sguard
.
release
(),
p
};
return
{
fd
,
ntohs
(
serv_addr
.
sin_port
)};
}
}
std
::
pair
<
default_socket_acceptor
,
uint16_t
>
std
::
pair
<
default_socket_acceptor
,
uint16_t
>
new_
ipv4
_acceptor
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse
)
{
new_
tcp
_acceptor
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse
)
{
CAF_LOGF_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
CAF_LOGF_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
auto
&
backend
=
get_multiplexer_singleton
();
auto
&
backend
=
get_multiplexer_singleton
();
auto
acceptor
=
new_
ipv4
_acceptor_impl
(
port
,
addr
,
reuse
);
auto
acceptor
=
new_
tcp
_acceptor_impl
(
port
,
addr
,
reuse
);
auto
bound_port
=
acceptor
.
second
;
auto
bound_port
=
acceptor
.
second
;
CAF_REQUIRE
(
port
==
0
||
bound_port
==
port
);
CAF_REQUIRE
(
port
==
0
||
bound_port
==
port
);
return
{
default_socket_acceptor
{
backend
,
std
::
move
(
acceptor
.
first
)},
return
{
default_socket_acceptor
{
backend
,
std
::
move
(
acceptor
.
first
)},
...
...
libcaf_io/src/interfaces.cpp
View file @
aae2e2b9
...
@@ -19,8 +19,8 @@
...
@@ -19,8 +19,8 @@
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/interfaces.hpp"
#include <errno.h>
#include <netdb.h>
#include <netdb.h>
#include <errno.h>
#include <stdlib.h>
#include <stdlib.h>
#include <string.h>
#include <string.h>
#include <net/if.h>
#include <net/if.h>
...
@@ -144,6 +144,39 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
...
@@ -144,6 +144,39 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
return
result
;
return
result
;
}
}
optional
<
std
::
pair
<
std
::
string
,
protocol
>>
interfaces
::
native_address
(
const
std
::
string
&
host
,
optional
<
protocol
>
preferred
)
{
addrinfo
hint
;
memset
(
&
hint
,
0
,
sizeof
(
hint
));
hint
.
ai_socktype
=
SOCK_STREAM
;
if
(
preferred
)
{
hint
.
ai_family
=
*
preferred
==
protocol
::
ipv4
?
AF_INET
:
AF_INET6
;
}
addrinfo
*
tmp
=
nullptr
;
if
(
getaddrinfo
(
host
.
c_str
(),
nullptr
,
&
hint
,
&
tmp
))
{
return
none
;
}
std
::
unique_ptr
<
addrinfo
,
decltype
(
freeaddrinfo
)
*>
addrs
{
tmp
,
freeaddrinfo
};
for
(
auto
i
=
addrs
.
get
();
i
!=
nullptr
;
i
=
i
->
ai_next
)
{
auto
family
=
i
->
ai_family
;
if
(
family
==
AF_INET
||
family
==
AF_INET6
)
{
char
buffer
[
INET6_ADDRSTRLEN
];
auto
res
=
family
==
AF_INET
?
inet_ntop
(
family
,
&
reinterpret_cast
<
sockaddr_in
*>
(
i
->
ai_addr
)
->
sin_addr
,
buffer
,
sizeof
(
buffer
))
:
inet_ntop
(
family
,
&
reinterpret_cast
<
sockaddr_in6
*>
(
i
->
ai_addr
)
->
sin6_addr
,
buffer
,
sizeof
(
buffer
));
if
(
res
!=
nullptr
)
{
return
{{
res
,
family
==
AF_INET
?
protocol
::
ipv4
:
protocol
::
ipv6
}};
}
}
}
return
none
;
}
}
// namespace network
}
// namespace network
}
// namespace io
}
// namespace io
}
// namespace caf
}
// namespace caf
unit_testing/test_unpublish.cpp
View file @
aae2e2b9
...
@@ -65,6 +65,7 @@ int main() {
...
@@ -65,6 +65,7 @@ int main() {
auto
port
=
publish_at_some_port
(
4242
,
d
);
auto
port
=
publish_at_some_port
(
4242
,
d
);
std
::
this_thread
::
sleep_for
(
std
::
chrono
::
milliseconds
(
50
));
std
::
this_thread
::
sleep_for
(
std
::
chrono
::
milliseconds
(
50
));
io
::
unpublish
(
d
,
port
);
io
::
unpublish
(
d
,
port
);
CAF_CHECKPOINT
();
// must fail now
// must fail now
try
{
try
{
auto
oops
=
io
::
remote_actor
(
"127.0.0.1"
,
port
);
auto
oops
=
io
::
remote_actor
(
"127.0.0.1"
,
port
);
...
...
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