Commit a1a11c74 authored by Dominik Charousset's avatar Dominik Charousset

cleaned up MinGW changes, removed node_id::get

this patch streamlines the recent patch from ajac: enforce consistent
coding style, removed the superfluous windows_tcp class and moved
the node_id::get to the MM, to guarantee WSAStartup has been called
before accessing network devices
parent 5727f72b
......@@ -116,7 +116,7 @@ if (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
elseif (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
set(LIBCPPA_PLATFORM_SRC src/middleman_event_handler_poll.cpp)
elseif (${CMAKE_SYSTEM_NAME} MATCHES "Window")
set(LIBCPPA_PLATFORM_SRC src/middleman_event_handler_poll.cpp src/backtrace.cpp)
set(LIBCPPA_PLATFORM_SRC src/middleman_event_handler_poll.cpp src/execinfo_windows.cpp)
else ()
message(FATAL_ERROR "This platform is not supported by libcppa")
endif()
......@@ -129,8 +129,7 @@ file(GLOB LIBCPPA_HDRS "cppa/*.hpp"
"cppa/io/*.hpp"
"cppa/opencl/*.hpp"
"cppa/qtsupport/*.hpp"
"cppa/util/*.hpp"
"cppa/windows/*.hpp")
"cppa/util/*.hpp")
# list cpp files including platform-dependent files
set(LIBCPPA_SRC
......@@ -206,7 +205,6 @@ set(LIBCPPA_SRC
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
src/socketpair.cpp
src/string_serialization.cpp
src/sync_request_bouncer.cpp
src/thread_pool_scheduler.cpp
......@@ -248,13 +246,6 @@ if (ENABLE_OPENCL)
add_definitions(-DCPPA_OPENCL)
endif (ENABLE_OPENCL)
if (MINGW)
set(LIBCPPA_SRC
${LIBCPPA_SRC}
src/windows/windows_tcp.cpp)
endif (MINGW)
if (DISABLE_MEM_MANAGEMENT)
add_definitions(-DCPPA_DISABLE_MEM_MANAGEMENT)
endif (DISABLE_MEM_MANAGEMENT)
......
......@@ -351,3 +351,5 @@ unit_testing/test_yield_interface.cpp
cppa/may_have_timeout.hpp
examples/message_passing/typed_calculator.cpp
examples/aout.cpp
cppa/detail/execinfo_windows.hpp
src/execinfo_windows.cpp
......@@ -31,6 +31,7 @@
#ifndef CPPA_CONFIG_HPP
#define CPPA_CONFIG_HPP
// detect compiler and set CPPA_DEPRECATED
#if defined(__clang__)
# define CPPA_CLANG
# define CPPA_DEPRECATED __attribute__((__deprecated__))
......@@ -43,6 +44,7 @@
# define CPPA_DEPRECATED
#endif
// detect OS
#if defined(__APPLE__)
# define CPPA_MACOS
# ifndef _GLIBCXX_HAS_GTHREADS
......@@ -64,29 +66,36 @@
#include <cstdio>
#include <cstdlib>
#ifdef CPPA_DEBUG_MODE
// import backtrace and backtrace_symbols_fd into cppa::detail
#ifdef CPPA_WINDOWS
#include <cppa/execinfo_windows.h>
#include "cppa/detail/execinfo_windows.hpp"
#else
#include <execinfo.h>
namespace cppa {
namespace detail {
using ::backtrace;
using ::backtrace_symbols_fd;
} // namespace detail
} // namespace cppa
#endif
#define CPPA_REQUIRE__(stmt, file, line) \
//
#ifdef CPPA_DEBUG_MODE
# define CPPA_REQUIRE__(stmt, file, line) \
printf("%s:%u: requirement failed '%s'\n", file, line, stmt); \
{ \
void *array[10]; \
size_t size = backtrace(array, 10); \
backtrace_symbols_fd(array, size, 2); \
void* array[10]; \
size_t size = ::cppa::detail::backtrace(array, 10); \
::cppa::detail::backtrace_symbols_fd(array, size, 2); \
} \
abort()
#define CPPA_REQUIRE(stmt) \
# define CPPA_REQUIRE(stmt) \
if ((stmt) == false) { \
CPPA_REQUIRE__(#stmt, __FILE__, __LINE__); \
}((void) 0)
#else // CPPA_DEBUG_MODE
#define CPPA_REQUIRE(unused) static_cast<void>(0)
#endif // CPPA_DEBUG_MODE
#else
# define CPPA_REQUIRE(unused) static_cast<void>(0)
#endif
#define CPPA_CRITICAL__(error, file, line) { \
printf("%s:%u: critical error: '%s'\n", file, line, error); \
......@@ -97,17 +106,18 @@
#ifdef CPPA_WINDOWS
# include <w32api.h>
#undef _WIN32_WINNT
#undef WINVER
#define _WIN32_WINNT WindowsVista
#define WINVER WindowsVista
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <ws2tcpip.h>
# include <winsock2.h>
#undef interface
// remove interface which is defined in rpc.h in files included by windows.h
// as it clashes with name used in own code
// remove interface which is defined in rpc.h in files included by
// windows.h as it clashes with name used in own code
# undef interface
#else
# include <unistd.h>
# include <errno.h>
#endif
namespace cppa {
......@@ -120,17 +130,29 @@ std::unique_ptr<T> create_unique(Args&&... args) {
return std::unique_ptr<T>{new T(std::forward<Args>(args)...)};
}
// platform-dependent types for sockets and some utility functions
#ifdef CPPA_WINDOWS
typedef SOCKET native_socket_type;
typedef const char* setsockopt_ptr;
typedef const char* socket_send_ptr;
typedef char* socket_recv_ptr;
typedef int socklen_t;
constexpr SOCKET invalid_socket = INVALID_SOCKET;
inline int last_socket_error() { return WSAGetLastError(); }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode = WSATRY_AGAIN;
}
#else
typedef int native_socket_type;
typedef const void* setsockopt_ptr;
typedef const void* socket_send_ptr;
typedef void* socket_recv_ptr;
constexpr int invalid_socket = -1;
inline void closesocket(native_socket_type fd) { close(fd); }
inline int last_socket_error() { return errno; }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
}
#endif
} // namespace cppa
......
......@@ -11,12 +11,11 @@
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* Raphael Hiesgen <raphael.hiesgen@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
......@@ -28,44 +27,45 @@
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <atomic>
#include <iostream>
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
#include "cppa/windows/windows_tcp.hpp"
using namespace std;
namespace cppa { namespace windows {
void windows_tcp::initialize()
{
WSADATA WinsockData;
if (WSAStartup(MAKEWORD(2, 2), &WinsockData) != 0) {
ostringstream oss;
oss << "libcppa WSAStartup failed. ";
CPPA_LOGMF(CPPA_ERROR, oss.str());
throw runtime_error(oss.str());
}
}
/******************************************************************************\
* Based on work by the mingw-w64 project; *
* original header: *
* *
* Copyright (c) 2012 mingw-w64 project *
* *
* Contributing author: Kai Tietz *
* *
* Permission is hereby granted, free of charge, to any person obtaining a *
* copy of this software and associated documentation files (the "Software"), *
* to deal in the Software without restriction, including without limitation *
* the rights to use, copy, modify, merge, publish, distribute, sublicense, *
* and/or sell copies of the Software, and to permit persons to whom the *
* Software is furnished to do so, subject to the following conditions: *
* *
* The above copyright notice and this permission notice shall be included in *
* all copies or substantial portions of the Software. *
* *
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *
* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *
* DEALINGS IN THE SOFTWARE. *
\******************************************************************************/
void windows_tcp::destroy() {
WSACleanup();
delete this;
}
#ifndef CPPA_DETAIL_EXECINFO_WINDOWS_HPP
#define CPPA_DETAIL_EXECINFO_WINDOWS_HPP
void windows_tcp::dispose() {
WSACleanup();
delete this;
}
namespace cppa {
namespace detail {
windows_tcp* get_windows_tcp() {
return detail::singleton_manager::get_windows_tcp();
}
int backtrace(void** buffer, int size);
char** backtrace_symbols(void* const* buffer, int size);
void backtrace_symbols_fd(void* const* buffer, int size, int fd);
} } // namespace cppa::windows
} // namespace detail
} // namespace cppa
#endif // CPPA_DETAIL_EXECINFO_WINDOWS_HPP
......@@ -31,57 +31,24 @@
#ifndef FD_UTIL_HPP
#define FD_UTIL_HPP
#include <unistd.h>
#include <string>
#include <utility> // std::pair
#include <unistd.h> // ssize_t
#include "cppa/config.hpp"
namespace cppa { namespace detail { namespace fd_util {
#if defined(CPPA_MACOS) || defined(CPPA_LINUX)
// throws ios_base::failure and adds errno failure if @p add_errno_failure
void throw_io_failure(const char* what, bool add_errno_failure = true);
// returns true if fd is nonblocking
// throws @p ios_base::failure on error
bool nonblocking(native_socket_type fd);
// sets fd to nonblocking if <tt>set_nonblocking == true</tt>
// or to blocking if <tt>set_nonblocking == false</tt>
// throws @p ios_base::failure on error
void nonblocking(native_socket_type fd, bool new_value);
// returns true if fd is nodelay socket
// throws @p ios_base::failure on error
bool tcp_nodelay(native_socket_type fd);
// returns true if fd is nodelay socket
// throws @p ios_base::failure on error
void tcp_nodelay(native_socket_type fd, bool new_value);
// reads @p result and @p errno and throws @p ios_base::failure on error
void handle_write_result(ssize_t result, bool is_nonblocking_io);
// reads @p result and @p errno and throws @p ios_base::failure on error
void handle_read_result(ssize_t result, bool is_nonblocking_io);
#else
std::string last_socket_error_as_string();
// throws ios_base::failure and adds errno failure if @p add_errno_failure
void throw_io_failure(const char* what, bool add_errno_failure = true);
// returns true if fd is nonblocking
// throws @p ios_base::failure on error
bool nonblocking(native_socket_type fd);
// sets fd to nonblocking if <tt>set_nonblocking == true</tt>
// or to blocking if <tt>set_nonblocking == false</tt>
// throws @p ios_base::failure on error
void nonblocking(native_socket_type fd, bool new_value);
// returns true if fd is nodelay socket
// throws @p ios_base::failure on error
bool tcp_nodelay(native_socket_type fd);
// returns true if fd is nodelay socket
// throws @p ios_base::failure on error
void tcp_nodelay(native_socket_type fd, bool new_value);
......@@ -92,7 +59,7 @@ void handle_write_result(ssize_t result, bool is_nonblocking_io);
// reads @p result and @p errno and throws @p ios_base::failure on error
void handle_read_result(ssize_t result, bool is_nonblocking_io);
#endif
std::pair<native_socket_type, native_socket_type> create_pipe();
} } } // namespace cppa::detail::fd_util
......
......@@ -44,8 +44,6 @@ namespace cppa { namespace io { class middleman; } }
namespace cppa { namespace opencl { class opencl_metainfo; } }
namespace cppa { namespace windows { class windows_tcp; } }
namespace cppa { namespace detail {
class empty_tuple;
......@@ -82,8 +80,6 @@ class singleton_manager {
static opencl::opencl_metainfo* get_opencl_metainfo();
static windows::windows_tcp* get_windows_tcp();
private:
/*
......
/*
Copyright (c) 2012 mingw-w64 project
Contributing author: Kai Tietz
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
*/
int backtrace (void **buffer, int size);
char ** backtrace_symbols (void *const *buffer, int size);
void backtrace_symbols_fd (void *const *buffer, int size, int fd);
......@@ -173,6 +173,11 @@ class middleman {
*/
inline actor_namespace& get_namespace();
/**
* @brief Returns the node of this middleman.
*/
inline const node_id_ptr& node() const;
protected:
// creates a middleman instance
......@@ -197,12 +202,14 @@ class middleman {
};
int dumb_socketpair(native_socket_type socks[2], int make_overlapped);
inline actor_namespace& middleman::get_namespace() {
return m_namespace;
}
const node_id_ptr& middleman::node() const {
return m_node;
}
} } // namespace cppa::io
#endif // MIDDLEMAN_HPP
......@@ -34,14 +34,10 @@
#include <cstring>
#include <sstream>
#include <iostream>
#include <cppa/config.hpp>
#ifdef CPPA_WINDOWS
#include <cppa/execinfo_windows.h>
#else
#include <execinfo.h>
#endif
#include <type_traits>
#include "cppa/config.hpp"
#include "cppa/singletons.hpp"
#include "cppa/abstract_actor.hpp"
......@@ -171,8 +167,8 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
std::cerr << "[" << lvlname << "] " << classname << "::" \
<< funname << ": " << message << "\nStack trace:\n"; \
void* bt_array[20]; \
size_t size = backtrace(bt_array, 20); \
backtrace_symbols_fd(bt_array, size, 2); \
size_t size = ::cppa::detail::backtrace(bt_array, 20); \
::cppa::detail::backtrace_symbols_fd(bt_array, size, 2); \
} CPPA_VOID_STMT
#ifndef CPPA_LOG_LEVEL
......
......@@ -94,12 +94,6 @@ class node_id : public ref_counted, util::comparable<node_id> {
*/
inline const host_id_type& host_id() const { return m_host_id; }
/**
* @brief Returns the proccess_information for the running process.
* @returns A pointer to the singleton of this process.
*/
static const intrusive_ptr<node_id>& get();
/** @cond PRIVATE */
// "inherited" from comparable<node_id>
......
......@@ -67,13 +67,6 @@ inline detail::empty_tuple* get_empty_tuple() {
return detail::singleton_manager::get_empty_tuple();
}
#ifdef CPPA_WINDOWS
inline windows::windows_tcp* get_windows_tcp() {
return detail::singleton_manager::get_windows_tcp();
}
#endif
} // namespace cppa
#endif // CPPA_SINGLETONS_HPP
......@@ -127,11 +127,8 @@ int main(int argc, char** argv) {
}
},
on("/quit") >> [&] {
#ifndef CPPA_WINDOWS
close(STDIN_FILENO); // close STDIN; causes this match loop to quit
#else
cin.setstate(ios_base::eofbit); // close STDIN; causes this match loop to quit
#endif
// close STDIN; causes this match loop to quit
cin.setstate(ios_base::eofbit);
},
on<string, anything>().when(_x1.starts_with("/")) >> [&] {
cout << "*** available commands:\n"
......
......@@ -29,9 +29,12 @@
#include "cppa/actor.hpp"
#include "cppa/singletons.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/raw_access.hpp"
namespace cppa {
......@@ -64,7 +67,7 @@ actor_id actor_addr::id() const {
}
const node_id& actor_addr::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
return m_ptr ? m_ptr->node() : *get_middleman()->node();
}
bool actor_addr::is_remote() const {
......
......@@ -72,7 +72,7 @@ actor_addr actor_namespace::read(deserializer* source) {
auto aid = source->read<uint32_t>(); // actor id
auto pid = source->read<uint32_t>(); // process id
source->read_raw(node_id::host_id_size, hid.data()); // host id
auto this_node = node_id::get();
auto this_node = get_middleman()->node();
if (aid == 0 && pid == 0) {
// 0:0 identifies an invalid actor
return invalid_actor_addr;
......
......@@ -35,9 +35,12 @@
#include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -46,7 +49,7 @@ using namespace std;
namespace cppa {
actor_proxy::actor_proxy(actor_id mid) : super(mid) {
m_node = node_id::get();
m_node = get_middleman()->node();
}
} // namespace cppa
/*
Copyright (c) 2012 mingw-w64 project
Contributing author: Kai Tietz
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
*/
#include <windows.h>
/*
#include <dbghelp.h>
*/
#include <stdio.h>
#include <stdlib.h>
#include <io.h>
int backtrace (void **buffer, int size);
char ** backtrace_symbols (void *const *buffer, int size);
void backtrace_symbols_fd (void *const *buffer, int size, int fd);
int
backtrace (void **buffer, int size)
{
USHORT frames;
HANDLE hProcess;
if (size <= 0)
return 0;
hProcess = GetCurrentProcess ();
frames = CaptureStackBackTrace (0, (DWORD) size, buffer, NULL);
return (int) frames;
}
char **
backtrace_symbols (void *const *buffer, int size)
{
size_t t = 0;
char **r, *cur;
int i;
t = ((size_t) size) * ((sizeof (void *) * 2) + 6 + sizeof (char *));
r = (char **) malloc (t);
if (!r)
return r;
cur = (char *) &r[size];
for (i = 0; i < size; i++)
{
r[i] = cur;
sprintf (cur, "[+0x%Ix]", (size_t) buffer[i]);
cur += strlen (cur) + 1;
}
return r;
}
void
backtrace_symbols_fd (void *const *buffer, int size, int fd)
{
char s[128];
int i;
for (i = 0; i < size; i++)
{
sprintf (s, "[+0x%Ix]\n", (size_t) buffer[i]);
write (fd, s, strlen (s));
}
_commit (fd);
}
/*
void printStack (void);
void printStack (void)
{
unsigned int i;
void *stack[100];
unsigned short frames;
SYMBOL_INFO *symbol;
HANDLE hProcess;
hProcess = GetCurrentProcess ();
SymInitialize (hProcess, NULL, TRUE);
frames = CaptureStackBackTrace( 0, 100, stack, NULL );
symbol = (SYMBOL_INFO *) calloc (sizeof (SYMBOL_INFO) + 256 * sizeof (char), 1);
symbol->MaxNameLen = 255;
symbol->SizeOfStruct = sizeof (SYMBOL_INFO);
for (i = 0;i < frames; i++)
{
SymFromAddr (hProcess, (DWORD_PTR) (stack[i]), 0, symbol);
printf ("%u: %p %s = 0x%Ix\n", frames - i - 1, stack[i], symbol->Name, symbol->Address);
}
free (symbol);
}
int main ()
{
printStack ();
return 0;
}
*/
......@@ -29,19 +29,20 @@
#include <sstream>
#include <stdlib.h>
#include "cppa/config.hpp"
#include "cppa/exception.hpp"
#ifndef CPPA_WINDOWS
#include <sys/socket.h>
#include <sys/un.h>
#ifdef CPPA_WINDOWS
# include <winerror.h>
#else
#include <winerror.h>
# include <errno.h>
# include <sys/socket.h>
# include <sys/un.h>
#endif
#include <stdlib.h>
#include <errno.h>
namespace {
std::string ae_what(std::uint32_t reason) {
......@@ -52,19 +53,19 @@ std::string ae_what(std::uint32_t reason) {
std::string be_what(int err_code) {
switch (err_code) {
#ifndef CPPA_WINDOWS
case EACCES: return "EACCES: address is protected; root access needed";
case EADDRINUSE: return "EADDRINUSE: address is already in use";
# ifndef CPPA_WINDOWS
case EACCES: return "EACCES: address protected; root access needed";
case EADDRINUSE: return "EADDRINUSE: address already in use";
case EBADF: return "EBADF: no valid socket descriptor";
case EINVAL: return "EINVAL: socket already bound to an address";
case ENOTSOCK: return "ENOTSOCK: file descriptor given";
#else
case WSAEACCES: return "EACCES: address is protected; root access needed";
case WSAEADDRINUSE: return "EADDRINUSE: address is already in use";
case ENOTSOCK: return "ENOTSOCK: no file descriptor given";
# else
case WSAEACCES: return "EACCES: address protected; root access needed";
case WSAEADDRINUSE: return "EADDRINUSE: address already in use";
case WSAEBADF: return "EBADF: no valid socket descriptor";
case WSAEINVAL: return "EINVAL: socket already bound to an address";
case WSAENOTSOCK: return "ENOTSOCK: file descriptor given";
#endif
case WSAENOTSOCK: return "ENOTSOCK: no file descriptor given";
# endif
default: break;
}
std::stringstream oss;
......
......@@ -11,12 +11,11 @@
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* Raphael Hiesgen <raphael.hiesgen@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
......@@ -29,44 +28,72 @@
\******************************************************************************/
#ifndef WINDOWS_TCP_HPP
#define WINDOWS_TCP_HPP
#include <atomic>
#include <vector>
#include <algorithm>
#include <functional>
#include "cppa/cppa.hpp"
#include <winsock2.h>
#include <ws2tcpip.h>
#include "cppa/windows/windows_tcp.hpp"
#include "cppa/detail/singleton_mixin.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa { namespace windows {
class windows_tcp {
friend class detail::singleton_manager;
/******************************************************************************\
* Based on work by the mingw-w64 project; *
* original header: *
* *
* Copyright (c) 2012 mingw-w64 project *
* *
* Contributing author: Kai Tietz *
* *
* Permission is hereby granted, free of charge, to any person obtaining a *
* copy of this software and associated documentation files (the "Software"), *
* to deal in the Software without restriction, including without limitation *
* the rights to use, copy, modify, merge, publish, distribute, sublicense, *
* and/or sell copies of the Software, and to permit persons to whom the *
* Software is furnished to do so, subject to the following conditions: *
* *
* The above copyright notice and this permission notice shall be included in *
* all copies or substantial portions of the Software. *
* *
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *
* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *
* DEALINGS IN THE SOFTWARE. *
\******************************************************************************/
private:
static inline windows_tcp* create_singleton() {
return new windows_tcp;
#include <windows.h>
#include <stdio.h>
#include <stdlib.h>
#include <io.h>
#include "cppa/detail/execinfo_windows.hpp"
namespace cppa {
namespace detail {
int backtrace(void** buffer, int size) {
if (size <= 0) return 0;
auto hProcess = GetCurrentProcess();
auto frames = CaptureStackBackTrace (0, (DWORD) size, buffer, NULL);
return static_cast<int>(frames);
}
char** backtrace_symbols(void* const* buffer, int size) {
auto t = ((size_t) size) * ((sizeof(void*) * 2) + 6 + sizeof(char*));
auto r = (char**) malloc(t);
if (!r) return nullptr;
auto cur = (char *) &r[size];
for (int i = 0; i < size; i++) {
r[i] = cur;
sprintf (cur, "[+0x%Ix]", (size_t) buffer[i]);
cur += strlen (cur) + 1;
}
return r;
}
void backtrace_symbols_fd(void* const* buffer, int size, int fd) {
char s[128];
for (int i = 0; i < size; i++) {
sprintf(s, "[+0x%Ix]\n", (size_t) buffer[i]);
write(fd, s, strlen (s));
}
_commit(fd);
}
void initialize();
void dispose();
void destroy();
};
windows_tcp* get_windows_tcp();
} } // namespace cppa::windows
#endif // WINDOWS_TCP_HPP
} // namespace detail
} // namespace cppa
......@@ -30,61 +30,42 @@
#include <ios>
#include <sstream>
#include <cstring>
// socket related functions
#include <sys/types.h>
#include <sys/socket.h>
#ifdef CPPA_WINDOWS
# include <winsock2.h>
# include <ws2tcpip.h> /* socklen_t, et al (MSVC20xx) */
# include <windows.h>
# include <io.h>
#else
# include <errno.h>
# include <netdb.h>
# include <fcntl.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
#endif
#include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp"
#if defined(CPPA_LINUX) || defined(CPPA_MACOS)
#include <errno.h>
#include <netdb.h>
#include <fcntl.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
/************** implementation of platform-independent functions **************/
namespace cppa { namespace detail { namespace fd_util {
void throw_io_failure(const char* what, bool add_errno_failure) {
if (add_errno_failure) {
std::ostringstream oss;
oss << what << ": " << strerror(errno)
<< " [errno: " << errno << "]";
oss << what << ": " << last_socket_error_as_string()
<< " [errno: " << last_socket_error() << "]";
throw std::ios_base::failure(oss.str());
}
throw std::ios_base::failure(what);
}
int rd_flags(native_socket_type fd) {
auto flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) throw_io_failure("unable to read socket flags");
return flags;
}
bool nonblocking(native_socket_type fd) {
return (rd_flags(fd) & O_NONBLOCK) != 0;
}
void nonblocking(native_socket_type fd, bool new_value) {
auto rf = rd_flags(fd);
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
if (fcntl(fd, F_SETFL, wf) < 0) {
throw_io_failure("unable to set file descriptor flags");
}
}
bool tcp_nodelay(native_socket_type fd) {
int flag;
auto len = static_cast<socklen_t>(sizeof(flag));
if (getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, &len) < 0) {
throw_io_failure("unable to read TCP_NODELAY socket option");
}
return flag != 0;
}
void tcp_nodelay(native_socket_type fd, bool new_value) {
int flag = new_value ? 1 : 0;
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)) < 0) {
......@@ -94,8 +75,11 @@ void tcp_nodelay(native_socket_type fd, bool new_value) {
void handle_io_result(ssize_t res, bool is_nonblock, const char* msg) {
if (res < 0) {
// don't throw for 'failed' non-blocking IO
if (is_nonblock && (errno == EAGAIN || errno == EWOULDBLOCK)) { }
auto err = last_socket_error();
if (is_nonblock && would_block_or_temporarily_unavailable(err)) {
// don't throw for 'failed' non-blocking IO,
// just try again later
}
else throw_io_failure(msg);
}
}
......@@ -111,74 +95,158 @@ void handle_read_result(ssize_t res, bool is_nonblock) {
} } } // namespace cppa::detail::fd_util
#else
// windows
#include <winsock2.h>
#ifndef CPPA_WINDOWS // Linux or Mac OS
namespace cppa { namespace detail { namespace fd_util {
void throw_io_failure(const char* what, bool add_errno_failure) {
if (add_errno_failure) {
std::ostringstream oss;
oss << what << ": " << strerror(errno)
<< " [errno: " << errno << "]";
throw std::ios_base::failure(oss.str());
}
throw std::ios_base::failure(what);
std::string last_socket_error_as_string() {
return strerror(errno);
}
// int rd_flags(native_socket_type fd) {
// auto flags = ioctlsocket(fd, F_GETFL, 0);
// if (flags == -1) throw_io_failure("unable to read socket flags");
// return flags;
// }
// bool nonblocking(native_socket_type fd) {
// return (rd_flags(fd) & O_NONBLOCK) != 0;
// }
void nonblocking(native_socket_type fd, bool new_value) {
u_long iMode = new_value ? 1 : 0;
// If iMode = 0, blocking is enabled;
// If iMode != 0, non-blocking mode is enabled.
if (ioctlsocket(fd, FIONBIO, &iMode) < 0) {
throw_io_failure("unable to set FIONBIO");
// read flags for fd
auto rf = fcntl(fd, F_GETFL, 0);
if (rf == -1) throw_io_failure("unable to read socket flags");
// calculate and set new flags
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
if (fcntl(fd, F_SETFL, wf) < 0) {
throw_io_failure("unable to set file descriptor flags");
}
}
bool tcp_nodelay(native_socket_type fd) {
int flag;
auto len = static_cast<int>(sizeof(flag));
if (getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char *> (&flag), &len) < 0) {
throw_io_failure("unable to read TCP_NODELAY socket option");
}
return flag != 0;
std::pair<native_socket_type, native_socket_type> create_pipe() {
native_socket_type pipefds[2];
if (pipe(pipefds) != 0) { CPPA_CRITICAL("cannot create pipe"); }
return {pipefds[0], pipefds[1]};
}
void tcp_nodelay(native_socket_type fd, bool new_value) {
int flag = new_value ? 1 : 0;
auto len = static_cast<int>(sizeof(flag));
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char *> (&flag), len) < 0) {
throw_io_failure("unable to set TCP_NODELAY");
} } } // namespace cppa::detail::fd_util
#else // CPPA_WINDOWS
namespace cppa { namespace detail { namespace fd_util {
std::string last_socket_error_as_string() {
LPTSTR errorText = NULL;
FormatMessage(// use system message tables to retrieve error text
FORMAT_MESSAGE_FROM_SYSTEM
// allocate buffer on local heap for error text
| FORMAT_MESSAGE_ALLOCATE_BUFFER
// Important! will fail otherwise, since we're not
// (and CANNOT) pass insertion parameters
| FORMAT_MESSAGE_IGNORE_INSERTS,
nullptr, // unused with FORMAT_MESSAGE_FROM_SYSTEM
hresult,
MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) &errorText, // output
0, // minimum size for output buffer
nullptr); // arguments - see note
std::string result;
if (errorText != nullptr) {
result = errorText;
// release memory allocated by FormatMessage()
LocalFree(errorText);
}
return result;
}
void handle_io_result(ssize_t res, bool is_nonblock, const char* msg) {
if (res < 0) {
// don't throw for 'failed' non-blocking IO
if (is_nonblock && ( WSAGetLastError() == WSAEWOULDBLOCK)) { }
else throw_io_failure(msg);
void nonblocking(native_socket_type fd, bool new_value) {
u_long mode = new_value ? 1 : 0;
if (ioctlsocket(fd, FIONBIO, &mode) < 0) {
throw_io_failure("unable to set FIONBIO");
}
}
void handle_write_result(ssize_t res, bool is_nonblock) {
handle_io_result(res, is_nonblock, "cannot write to file descriptor");
// wraps a C call and throws an IO failure if f returns an integer != 0
template<typename F, typename... Ts>
inline void ccall(const char* errmsg, F f, Ts&&... args) {
if (f(std::forward<Ts>(args...)) != 0) {
throw_io_failure(errmsg);
}
}
void handle_read_result(ssize_t res, bool is_nonblock) {
handle_io_result(res, is_nonblock, "cannot read from file descriptor");
if (res == 0) throw_io_failure("cannot read from closed file descriptor");
/******************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions are met:*
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright notice, *
* this list of conditions and the following disclaimer in the documentation *
* and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED *
* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR *
* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR *
* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, *
* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, *
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;*
* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, *
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR *
* OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF *
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\******************************************************************************/
std::pair<native_socket_type, native_socket_type> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
native_socket_type socks[2] = {-1, -1};
auto listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == invalid_socket) throw_io_failure("socket() failed");
union {
sockaddr_in inaddr;
sockaddr addr;
} a;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
bool success = false;
// makes sure all sockets are closed in case of an error
auto guard = util::make_scope_guard([&] {
if (success) return; // everyhting's fine
auto e = WSAGetLastError();
closesocket(listener);
closesocket(socks[0]);
closesocket(socks[1]);
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
ccall("setsockopt() failed", ::setsockopt, listener,
SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&reuse),
static_cast<socklen_t>(sizeof(reuse)));
ccall("bind() failed", ::bind, listener, &a.addr, sizeof(a.inaddr));
// read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx):
memset(&a, 0, sizeof(a));
ccall("getsockname() failed", ::getsockname, listener, &a.addr, &addrlen);
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
ccall("listen() failed", ::listen, listener, 1);
// create read-only end of the pipe
DWORD flags = 0;
auto read_fd = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
if (read_fd == invalid_socket) {
throw_io_failure("cannot create read handle: WSASocket() failed");
}
ccall("connect() failed", ::connect, read_fd, &a.addr, sizeof(a.inaddr));
// get write-only end of the pipe
auto write_fd = accept(listener, NULL, NULL);
if (write_fd == invalid_socket) {
throw_io_failure("cannot create write handle: accept() failed");
}
closesocket(listener);
success = true;
return {read_fd, write_fd};
}
} } } // namespace cppa::detail::fd_util
......
......@@ -185,13 +185,19 @@ std::vector<std::string> get_mac_addresses() {
// Link with Iphlpapi.lib
// #pragma comment(lib, "IPHLPAPI.lib") -liphlpapi
#define WORKING_BUFFER_SIZE 15000
#define MAX_TRIES 3
namespace {
#define MALLOC(x) HeapAlloc(GetProcessHeap(), 0, (x))
#define FREE(x) HeapFree(GetProcessHeap(), 0, (x))
constexpr size_t working_buffer_size = 15 * 1024; // 15kb by default
constexpr size_t max_iterations = 3;
/* Note: could also use malloc() and free() */
struct c_free {
template<typename T>
void operator()(T* ptr) {
free(ptr);
}
};
} // namespace <anonymous>
using namespace std;
......@@ -199,95 +205,55 @@ using namespace std;
namespace cppa { namespace util {
std::vector<std::string> get_mac_addresses() {
/* Declare and initialize variables */
DWORD dwRetVal = 0;
unsigned int i = 0;
// Set the flags to pass to GetAdaptersAddresses
// result vector
vector<string> hw_addresses;
// flags to pass to GetAdaptersAddresses
ULONG flags = GAA_FLAG_INCLUDE_PREFIX;
// default to unspecified address family (both)
ULONG family = AF_UNSPEC;
PIP_ADAPTER_ADDRESSES pAddresses = NULL;
ULONG outBufLen = 0;
ULONG Iterations = 0;
PIP_ADAPTER_ADDRESSES pCurrAddresses = NULL;
vector<string> hw_addresses;
// Allocate a 15 KB buffer to start with.
outBufLen = WORKING_BUFFER_SIZE;
cppa::windows::get_windows_tcp();
// buffer
std::unique_ptr<IP_ADAPTER_ADDRESSES, c_free> addresses;
// init buf size to default, adjusted by GetAdaptersAddresses if needed
ULONG addresses_len = working_buffer_size;
// stores result of latest system call
DWORD res = 0;
// break condition
size_t iterations = 0;
do {
pAddresses = (IP_ADAPTER_ADDRESSES *) MALLOC(outBufLen);
if (pAddresses == NULL) {
addresses.reset((IP_ADAPTER_ADDRESSES*) malloc(addresses_len));
if (!addresses) {
perror("Memory allocation failed for IP_ADAPTER_ADDRESSES struct");
exit(1);
}
dwRetVal =
GetAdaptersAddresses(family, flags, NULL, pAddresses, &outBufLen);
if (dwRetVal == ERROR_BUFFER_OVERFLOW) {
FREE(pAddresses);
pAddresses = NULL;
} else {
break;
}
Iterations++;
} while ((dwRetVal == ERROR_BUFFER_OVERFLOW) && (Iterations < MAX_TRIES));
if (dwRetVal == NO_ERROR) {
// If successful, output some information from the data we received
pCurrAddresses = pAddresses;
while (pCurrAddresses) {
if (pCurrAddresses->PhysicalAddressLength != 0) {
res = GetAdaptersAddresses(family, flags, nullptr,
addresses.get(), &addresses_len);
} while ((res == ERROR_BUFFER_OVERFLOW) && (++iterations < max_iterations));
if (res == NO_ERROR) {
// read hardware addresses from the output we've received
for (auto addr = addresses.get(); addr != nullptr; addr = addr->Next) {
if (addr->PhysicalAddressLength > 0) {
std::ostringstream oss;
oss << hex;
for (i = 0; i < (unsigned int) pCurrAddresses->PhysicalAddressLength; i++) {
if (i == (pCurrAddresses->PhysicalAddressLength - 1)) {
oss.width(2);
oss << ((int) pCurrAddresses->PhysicalAddress[i]);
} else {
oss << static_cast<int>(addr->PhysicalAddress[0]);
for (DWORD i = 1; i < addr->PhysicalAddressLength; ++i) {
oss << ":";
oss.width(2);
oss << ((int) pCurrAddresses->PhysicalAddress[i]) << ":";
oss << static_cast<int>(addr->PhysicalAddress[i]);
}
auto hw_addr = oss.str();
if (hw_addr != "00:00:00:00:00:00") {
hw_addresses.push_back(std::move(hw_addr));
}
auto addr = oss.str();
if (addr != "00:00:00:00:00:00") {
hw_addresses.push_back(std::move(addr));
}
}
pCurrAddresses = pCurrAddresses->Next;
}
} else {
if (dwRetVal == ERROR_NO_DATA) {
if (res == ERROR_NO_DATA) {
perror("No addresses were found for the requested parameters");
} else {
perror("Call to GetAdaptersAddresses failed with error");
}
}
if (pAddresses) {
FREE(pAddresses);
}
return hw_addresses;
}
......
......@@ -31,6 +31,10 @@
#include "cppa/config.hpp"
#include "cppa/util/get_root_uuid.hpp"
namespace {
constexpr char uuid_format[] = "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF";
} // namespace <anonmyous>
#ifdef CPPA_MACOS
namespace {
......@@ -174,7 +178,7 @@ std::string get_root_uuid() {
auto cpy = uuid;
replace_if(cpy.begin(), cpy.end(), ::isxdigit, 'F');
// discard invalid UUID
if (cpy != "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF") uuid.clear();
if (cpy != uuid_format) uuid.clear();
// "\\?\Volume{5ec70abf-058c-11e1-bdda-806e6f6e6963}\"
}
......@@ -185,62 +189,42 @@ std::string get_root_uuid() {
#elif defined(CPPA_WINDOWS)
#include <string>
#include <windows.h>
#include <stdio.h>
#include <tchar.h>
#include <algorithm>
#include <string>
#include <iostream>
#include <algorithm>
using namespace std;
namespace cppa { namespace util {
namespace { constexpr size_t max_drive_name = MAX_PATH; }
std::string get_root_uuid() {
// what should be done here ??
string uuid;
#define BUFSIZE MAX_PATH
BOOL bFlag;
TCHAR Buf[BUFSIZE]; // temporary buffer for volume name
TCHAR Drive[] = TEXT("c:\\"); // template drive specifier
TCHAR I; // generic loop counter
// Walk through legal drive letters, skipping floppies.
for (I = TEXT('c'); I < TEXT('z'); I++ )
{
char drive_name[max_drive_name]; // temporary buffer for volume name
auto drive = TEXT("c:\\"); // string "template" for drive specifier
// walk through legal drive letters, skipping floppies
for (auto i = TEXT('c'); i < TEXT('z'); i++ ) {
// Stamp the drive for the appropriate letter.
Drive[0] = I;
bFlag = GetVolumeNameForVolumeMountPoint(
Drive, // input volume mount point or directory
Buf, // output volume name buffer
BUFSIZE ); // size of volume name buffer
if (bFlag)
{
string str(Buf);
std::size_t founds = str.find("Volume{");
if( founds != std::string::npos ) {
founds = founds+7;
std::size_t founde = str.find("}");
if( founde != std::string::npos && (founde > founds)) {
uuid = str.substr(founds,founde-founds);
drive[0] = i;
if (GetVolumeNameForVolumeMountPoint(drive, drive_name, max_drive_name)) {
auto first = drive_name.find("Volume{");
if (first != std::string::npos) {
first += 7;
auto last = drive_name.find("}", first);
if (last != std::string::npos && last > first) {
uuid = drive_name.substr(first, last - first);
// UUIDs are formatted as 8-4-4-4-12 hex digits groups
auto cpy = uuid;
replace_if(cpy.begin(), cpy.end(), ::isxdigit, 'F');
// discard invalid UUID
if (cpy != "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF") {
uuid.clear();
} else {
return uuid;
}
if (cpy != uuid_format) uuid.clear();
else return uuid; // return first valid UUID we get
}
}
}
}
return uuid;
......
......@@ -290,7 +290,7 @@ class local_group_module : public abstract_group::module {
public:
local_group_module()
: super("local"), m_process(node_id::get())
: super("local"), m_process(get_middleman()->node())
, m_actor_utype(uniform_typeid<actor>()){ }
group get(const string& identifier) override {
......
......@@ -26,17 +26,13 @@
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/config.hpp"
#ifdef CPPA_WINDOWS
# include <ws2tcpip.h>
# include <winsock2.h>
#endif
#include <ios>
#include <cstring>
#include <errno.h>
#include <iostream>
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
......@@ -86,38 +82,48 @@ struct socket_guard {
};
#ifdef CPPA_WINDOWS
inline native_socket_type do_accept(native_socket_type fd,
sockaddr* addr,
socklen_t* addrlen) {
return ::WSAAccept(fd, addr, addrlen, 0, 0)
}
#else
inline native_socket_type do_accept(native_socket_type fd,
sockaddr* addr,
socklen_t* addrlen) {
return ::accept(fd, addr, addrlen);
}
#endif
bool accept_impl(stream_ptr_pair& result,
native_socket_type fd,
bool nonblocking) {
/*
sockaddr addr;
memset(&addr, 0, sizeof(addr));
*/
//socklen_t addrlen = sizeof(addr);
/*
socklen_t addrlen;
memset(&addrlen, 0, sizeof(addrlen));
# ifdef CPPA_WINDOWS
addrlen = sizeof(addr);
# endif
*/
sockaddr addr;
#ifdef CPPA_WINDOWS
int addrlen;
#else
socklen_t addrlen;
#endif
memset(&addr, 0, sizeof(addr));
#ifdef CPPA_WINDOWS
addrlen=sizeof(addr); // check this
auto sfd = ::WSAAccept(fd, &addr, &addrlen, 0, 0); // size too small
if (sfd == INVALID_SOCKET) {
if (nonblocking && (WSAGetLastError() == WSATRY_AGAIN || WSAGetLastError() == WSAEWOULDBLOCK)) {
// ok, try again
return false;
}
throw_io_failure("accept failed");
}
#else
memset(&addrlen, 0, sizeof(addrlen));
auto sfd = ::accept(fd, &addr, &addrlen);
if (sfd < 0) {
if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {
auto sfd = do_accept(fd, &addr, &addrlen);
if (sfd == invalid_socket) {
auto err = last_socket_error();
if (nonblocking && would_block_or_temporarily_unavailable(err)) {
// ok, try again
return false;
}
throw_io_failure("accept failed");
}
#endif
stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
result.first = ptr;
result.second = ptr;
......@@ -133,55 +139,32 @@ std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port,
const char* addr) {
CPPA_LOGM_TRACE("ipv4_acceptor", CPPA_ARG(port) << ", addr = "
<< (addr ? addr : "nullptr"));
native_socket_type sockfd;
#ifdef CPPA_WINDOWS
// ensure tcp has been initialized
cppa::windows::get_windows_tcp();
sockfd = INVALID_SOCKET;
sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sockfd == INVALID_SOCKET) {
throw network_error("could not create server socket");
}
#else
sockfd = socket(AF_INET, SOCK_STREAM, 0);
# ifdef CPPA_WINDOWS
// ensure that TCP has been initialized via WSAStartup
cppa::get_middleman();
# endif
native_socket_type sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) {
throw network_error("could not create server socket");
}
#endif
// sguard closes the socket in case of exception
socket_guard sguard(sockfd);
#ifdef CPPA_WINDOWS
int on = 1;
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char*)&on,sizeof(on)) < 0) {
throw_io_failure("unable to set SO_REUSEADDR");
}
#else
int on = 1;
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) < 0) {
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on), sizeof(on)) < 0) {
throw_io_failure("unable to set SO_REUSEADDR");
}
#endif
struct sockaddr_in serv_addr;
memset((char*) &serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
if (! addr) {
serv_addr.sin_addr.s_addr = INADDR_ANY;
}
#ifdef CPPA_WINDOWS
else if (::inet_pton(AF_INET, addr, &serv_addr.sin_addr) <= 0) {
throw network_error("invalid IPv4 address");
}
#else
else if (inet_pton(AF_INET, addr, &serv_addr.sin_addr) <= 0) {
throw network_error("invalid IPv4 address");
}
#endif
serv_addr.sin_port = htons(port);
if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0) {
if (bind(sockfd, (sockaddr*) &serv_addr, sizeof(serv_addr)) < 0) {
throw bind_failure(errno);
}
if (listen(sockfd, 10) != 0) {
......
......@@ -34,19 +34,14 @@
#include <iostream>
#include "cppa/config.hpp"
#ifdef CPPA_WINDOWS
# include <ws2tcpip.h>
# include <winsock2.h>
#endif
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
# include "cppa/singletons.hpp"
# include "cppa/windows/windows_tcp.hpp"
# include <ws2tcpip.h>
# include <winsock2.h>
#else
# include <netdb.h>
# include <unistd.h>
......@@ -56,9 +51,6 @@
# include <netinet/tcp.h>
#endif
namespace cppa { namespace io {
using namespace ::cppa::detail::fd_util;
......@@ -140,14 +132,12 @@ io::stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) {
CPPA_LOGF_TRACE(CPPA_ARG(host) << ", " << CPPA_ARG(port));
CPPA_LOGF_INFO("try to connect to " << host << " on port " << port);
struct sockaddr_in serv_addr;
struct hostent* server;
#ifdef CPPA_WINDOWS
// ensure tcp has been initialized
cppa::get_windows_tcp();
#endif
# ifdef CPPA_WINDOWS
// make sure TCP has been initialized via WSAStartup
cppa::get_middleman();
# endif
sockaddr_in serv_addr;
hostent* server;
native_socket_type fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == invalid_socket) {
throw network_error("socket creation failed");
......
......@@ -77,7 +77,7 @@ local_actor::local_actor()
, m_dummy_node(), m_current_node(&m_dummy_node)
, m_planned_exit_reason(exit_reason::not_exited)
, m_state(actor_state::ready) {
m_node = node_id::get();
m_node = get_middleman()->node();
}
local_actor::~local_actor() { }
......
......@@ -49,6 +49,10 @@
#include "cppa/binary_deserializer.hpp"
#include "cppa/util/buffer.hpp"
#include "cppa/util/algorithm.hpp"
#include "cppa/util/ripemd_160.hpp"
#include "cppa/util/get_root_uuid.hpp"
#include "cppa/util/get_mac_addresses.hpp"
#include "cppa/io/peer.hpp"
#include "cppa/io/acceptor.hpp"
......@@ -66,14 +70,48 @@
#include "cppa/intrusive/single_reader_queue.hpp"
#ifdef CPPA_WINDOWS
#include "io.h"
#include "fcntl.h"
# include <io.h>
# include <fcntl.h>
#endif
using namespace std;
namespace cppa { namespace io {
void notify_queue_event(native_socket_type fd) {
uint8_t dummy = 0;
// on unix, we have file handles, on windows, we actually have sockets
# ifdef CPPA_WINDOWS
auto res = ::send(fd, &dummy, sizeof(dummy), 0);
# else
auto res = ::write(fd, &dummy, sizeof(dummy));
# endif
// ignore result: an "error" means our middleman has been shut down
static_cast<void>(res);
}
size_t num_queue_events(native_socket_type fd) {
static constexpr size_t num_dummies = 64;
uint8_t dummies[num_dummies];
// on unix, we have file handles, on windows, we actually have sockets
# ifdef CPPA_WINDOWS
auto read_result = ::recv(fd, dummies, num_dummies, 0);
# else
auto read_result = ::read(fd, dummies, num_dummies);
# endif
if (read_result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// try again later
return 0;
}
else {
CPPA_LOGF_ERROR("cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe");
}
}
return static_cast<size_t>(read_result);
}
class middleman_event {
friend class intrusive::single_reader_queue<middleman_event>;
......@@ -132,31 +170,12 @@ class middleman_impl : public middleman {
public:
middleman_impl() : m_done(false) {
m_handler = middleman_event_handler::create();
m_namespace.set_proxy_factory([=](actor_id aid, node_id_ptr ptr) {
return make_counted<remote_actor_proxy>(aid, std::move(ptr), this);
});
m_namespace.set_new_element_callback([=](actor_id aid, const node_id& node) {
deliver(node,
{invalid_actor_addr, nullptr},
make_any_tuple(atom("MONITOR"),
node_id::get(),
aid));
});
}
middleman_impl() : m_done(false) { }
void run_later(function<void()> fun) override {
m_queue.enqueue(new middleman_event(move(fun)));
atomic_thread_fence(memory_order_seq_cst);
uint8_t dummy = 0;
// ignore result; write error only means middleman already exited
#ifdef CPPA_WINDOWS
auto res = ::send(m_pipe_write, (char *)&dummy, sizeof(dummy), 0);
#else
auto res = write(m_pipe_write, &dummy, sizeof(dummy));
#endif
static_cast<void>(res);
notify_queue_event(m_pipe_in);
}
bool register_peer(const node_id& node, peer* ptr) override {
......@@ -253,11 +272,11 @@ class middleman_impl : public middleman {
}
}
void register_acceptor(const actor_addr& whom, peer_acceptor* ptr) override {
void register_acceptor(const actor_addr& aa, peer_acceptor* ptr) override {
run_later([=] {
CPPA_LOGC_TRACE("cppa::io::middleman",
"register_acceptor$lambda", "");
m_acceptors[whom].push_back(ptr);
m_acceptors[aa].push_back(ptr);
continue_reader(ptr);
});
}
......@@ -265,19 +284,29 @@ class middleman_impl : public middleman {
protected:
void initialize() override {
#ifdef CPPA_WINDOWS
// if ( CreatePipe(&m_pipe_read,&m_pipe_write,0,4096) ) { CPPA_CRITICAL("cannot create pipe"); }
// DWORD dwMode = PIPE_NOWAIT;
// if ( !SetNamedPipeHandleState(&m_pipe_read,&dwMode,NULL,NULL) ) { CPPA_CRITICAL("cannot set PIPE_NOWAIT"); }
native_socket_type pipefds[2];
if ( cppa::io::dumb_socketpair(pipefds, 0) != 0) { CPPA_CRITICAL("cannot create pipe"); }
#else
int pipefds[2];
if (pipe(pipefds) != 0) { CPPA_CRITICAL("cannot create pipe"); }
#endif
m_pipe_read = pipefds[0];
m_pipe_write = pipefds[1];
detail::fd_util::nonblocking(m_pipe_read, true);
# ifdef CPPA_WINDOWS
WSADATA WinsockData;
if (WSAStartup(MAKEWORD(2, 2), &WinsockData) != 0) {
CPPA_CRITICAL("WSAStartup failed");
}
# endif
m_node = compute_node_id();
m_handler = middleman_event_handler::create();
m_namespace.set_proxy_factory([=](actor_id aid, node_id_ptr ptr) {
return make_counted<remote_actor_proxy>(aid, std::move(ptr), this);
});
m_namespace.set_new_element_callback([=](actor_id aid,
const node_id& node) {
deliver(node,
{invalid_actor_addr, nullptr},
make_any_tuple(atom("MONITOR"),
m_node,
aid));
});
auto pipefds = detail::fd_util::create_pipe();
m_pipe_out = pipefds.first;
m_pipe_in = pipefds.second;
detail::fd_util::nonblocking(m_pipe_out, true);
// start threads
m_thread = thread([this] { middleman_loop(this); });
}
......@@ -288,14 +317,25 @@ class middleman_impl : public middleman {
this->m_done = true;
});
m_thread.join();
closesocket(m_pipe_read);
closesocket(m_pipe_write);
#
closesocket(m_pipe_out);
closesocket(m_pipe_in);
# ifdef CPPA_WINDOWS
WSACleanup();
# endif
}
private:
static cppa::node_id_ptr compute_node_id() {
using namespace cppa::util;
auto macs = util::get_mac_addresses();
auto hd_serial_and_mac_addr = util::join(macs.begin(), macs.end())
+ util::get_root_uuid();
cppa::node_id::host_id_type node_id;
ripemd_160(node_id, hd_serial_and_mac_addr);
return new cppa::node_id(getpid(), node_id);
}
inline void quit() { m_done = true; }
inline bool done() const { return m_done; }
......@@ -305,9 +345,8 @@ class middleman_impl : public middleman {
middleman_event_handler& handler();
thread m_thread;
native_socket_type m_pipe_read;
native_socket_type m_pipe_write;
native_socket_type m_pipe_out;
native_socket_type m_pipe_in;
middleman_queue m_queue;
struct peer_entry {
......@@ -335,39 +374,12 @@ class middleman_overseer : public continuable {
continue_reading_result continue_reading() {
CPPA_LOG_TRACE("");
static constexpr size_t num_dummies = 64;
uint8_t dummies[num_dummies];
#ifdef CPPA_WINDOWS
auto read_result = ::recv(read_handle(), (char *)dummies, num_dummies, 0);
#else
auto read_result = ::read(read_handle(), dummies, num_dummies);
#endif
CPPA_LOG_DEBUG("read " << read_result << " messages from queue");
#ifdef CPPA_WINDOWS
if (read_result == SOCKET_ERROR) {
if (WSAGetLastError() == WSAEWOULDBLOCK) {
// try again later
return read_continue_later;
}
else {
CPPA_LOG_ERROR("cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe");
}
}
#else
if (read_result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// try again later
return read_continue_later;
}
else {
CPPA_LOG_ERROR("cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe");
}
}
#endif
for (int i = 0; i < read_result; ++i) {
// on MacOS, recv() on a pipe fd will fail,
// on Windows, our pipe is actually composed of two sockets
// and there's no read() function to read from sockets
auto events = num_queue_events(read_handle());
CPPA_LOG_DEBUG("read " << events << " messages from queue");
for (size_t i = 0; i < events; ++i) {
unique_ptr<middleman_event> msg(m_queue.try_pop());
if (!msg) {
CPPA_LOG_ERROR("nullptr dequeued");
......@@ -395,9 +407,10 @@ void middleman_loop(middleman_impl* impl) {
middleman_event_handler* handler = impl->m_handler.get();
CPPA_LOGF_TRACE("run middleman loop");
CPPA_LOGF_INFO("middleman runs at "
<< to_string(*node_id::get()));
<< to_string(impl->node()));
handler->init();
impl->continue_reader(new middleman_overseer(impl->m_pipe_read, impl->m_queue));
impl->continue_reader(new middleman_overseer(impl->m_pipe_out,
impl->m_queue));
handler->update();
while (!impl->done()) {
handler->poll([&](event_bitmask mask, continuable* io) {
......
......@@ -38,8 +38,11 @@
#include "cppa/config.hpp"
#include "cppa/node_id.hpp"
#include "cppa/serializer.hpp"
#include "cppa/singletons.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/util/algorithm.hpp"
#include "cppa/util/ripemd_160.hpp"
#include "cppa/util/get_root_uuid.hpp"
......@@ -47,27 +50,6 @@
namespace {
cppa::node_id* compute_proc_info() {
using namespace cppa::util;
auto macs = get_mac_addresses();
auto hd_serial_and_mac_addr = join(macs.begin(), macs.end())
+ get_root_uuid();
cppa::node_id::host_id_type node_id;
ripemd_160(node_id, hd_serial_and_mac_addr);
return new cppa::node_id(getpid(), node_id);
}
cppa::node_id_ptr s_pinfo;
struct pinfo_manager {
pinfo_manager() {
if (!s_pinfo) {
s_pinfo.reset(compute_proc_info());
}
}
}
s_pinfo_manager;
std::uint8_t hex_char_value(char c) {
if (isdigit(c)) {
return static_cast<std::uint8_t>(c - '0');
......@@ -146,10 +128,6 @@ std::string to_string(const node_id::host_id_type& node_id) {
return oss.str();
}
const intrusive_ptr<node_id>& node_id::get() {
return s_pinfo;
}
int node_id::compare(const node_id& other) const {
int tmp = strncmp(reinterpret_cast<const char*>(host_id().data()),
reinterpret_cast<const char*>(other.host_id().data()),
......
......@@ -112,7 +112,7 @@ continue_reading_result peer::continue_reading() {
memcpy(host_id.data(), m_rd_buf.offset_data(sizeof(uint32_t)),
node_id::host_id_size);
m_node.reset(new node_id(process_id, host_id));
if (*node_id::get() == *m_node) {
if (*parent()->node() == *m_node) {
std::cerr << "*** middleman warning: "
"incoming connection from self"
<< std::endl;
......@@ -203,7 +203,7 @@ void peer::monitor(const actor_addr&,
return;
}
auto entry = get_actor_registry()->get_entry(aid);
auto pself = node_id::get();
auto pself = parent()->node();
if (*node == *pself) {
CPPA_LOG_ERROR("received 'MONITOR' from pself");
......
......@@ -63,7 +63,7 @@ continue_reading_result peer_acceptor::continue_reading() {
}
if (opt) {
auto& pair = *opt;
auto& pself = node_id::get();
auto& pself = m_parent->node();
uint32_t process_id = pself->process_id();
try {
auto& out = pair.second;
......
......@@ -56,18 +56,6 @@ class opencl_metainfo : public detail::singleton_mixin<opencl_metainfo> { };
} } // namespace cppa::opencl
#endif
#ifdef CPPA_WINDOWS
# include "cppa/windows/windows_tcp.hpp"
#else
namespace cppa { namespace windows {
class windows_tcp : public detail::singleton_mixin<windows_tcp> { };
} } // namespace cppa::windows
#endif
namespace cppa { void shutdown() { detail::singleton_manager::shutdown(); } }
namespace cppa { namespace detail {
......@@ -82,7 +70,6 @@ std::atomic<group_manager*> s_group_manager;
std::atomic<empty_tuple*> s_empty_tuple;
std::atomic<scheduler*> s_scheduler;
std::atomic<logging*> s_logger;
std::atomic<windows::windows_tcp*> s_windows_tcp;
} // namespace <anonymous>
......@@ -104,11 +91,6 @@ void singleton_manager::shutdown() {
CPPA_LOGF_DEBUG("clear type info map");
destroy(s_uniform_type_info_map);
destroy(s_logger);
#ifdef CPPA_WINDOWS
CPPA_LOGF_DEBUG("WSACleanup");
destroy(s_windows_tcp);
#endif
}
opencl::opencl_metainfo* singleton_manager::get_opencl_metainfo() {
......@@ -120,15 +102,6 @@ opencl::opencl_metainfo* singleton_manager::get_opencl_metainfo() {
# endif
}
windows::windows_tcp* singleton_manager::get_windows_tcp() {
#ifdef CPPA_WINDOWS
return lazy_get(s_windows_tcp);
# else
CPPA_LOGF_ERROR("libcppa was compiled without windows support");
throw std::logic_error("libcppa was compiled without windows support");
# endif
}
actor_registry* singleton_manager::get_actor_registry() {
return lazy_get(s_actor_registry);
......
/* socketpair.c
Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org>
Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:
Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
The name of the author must not be used to endorse or promote products
derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
/* Changes:
* 2014-02-12: merge David Woodhouse, Ger Hobbelt improvements
* git.infradead.org/users/dwmw2/openconnect.git/commitdiff/bdeefa54
* github.com/GerHobbelt/selectable-socketpair
* always init the socks[] to -1/INVALID_SOCKET on error, both on Win32/64
* and UNIX/other platforms
* 2013-07-18: Change to BSD 3-clause license
* 2010-03-31:
* set addr to 127.0.0.1 because win32 getsockname does not always set it.
* 2010-02-25:
* set SO_REUSEADDR option to avoid leaking some windows resource.
* Windows System Error 10049, "Event ID 4226 TCP/IP has reached
* the security limit imposed on the number of concurrent TCP connect
* attempts." Bleah.
* 2007-04-25:
* preserve value of WSAGetLastError() on all error returns.
* 2007-04-22: (Thanks to Matthew Gregan <kinetik@flim.org>)
* s/EINVAL/WSAEINVAL/ fix trivial compile failure
* s/socket/WSASocket/ enable creation of sockets suitable as stdin/stdout
* of a child process.
* add argument make_overlapped
*/
#include <string.h>
#ifdef WIN32
# include <ws2tcpip.h> /* socklen_t, et al (MSVC20xx) */
# include <windows.h>
# include <io.h>
# include "cppa/singletons.hpp"
# include "cppa/windows/windows_tcp.hpp"
#else
# include <sys/types.h>
# include <sys/socket.h>
# include <errno.h>
#endif
namespace cppa { namespace io {
#ifdef WIN32
/* dumb_socketpair:
* If make_overlapped is nonzero, both sockets created will be usable for
* "overlapped" operations via WSASend etc. If make_overlapped is zero,
* socks[0] (only) will be usable with regular ReadFile etc., and thus
* suitable for use as stdin or stdout of a child process. Note that the
* sockets must be closed with closesocket() regardless.
*/
int dumb_socketpair(SOCKET socks[2], int make_overlapped)
{
union {
struct sockaddr_in inaddr;
struct sockaddr addr;
} a;
cppa::windows::get_windows_tcp();
SOCKET listener = INVALID_SOCKET;
int e;
socklen_t addrlen = sizeof(a.inaddr);
DWORD flags = (make_overlapped ? WSA_FLAG_OVERLAPPED : 0);
int reuse = 1;
if (socks == 0) {
WSASetLastError(WSAEINVAL);
return SOCKET_ERROR;
}
socks[0] = socks[1] = -1;
listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == INVALID_SOCKET) // -1
return SOCKET_ERROR;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
for (;;) {
if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
(char*) &reuse, (socklen_t) sizeof(reuse)) == -1)
break;
if (bind(listener, &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
break;
memset(&a, 0, sizeof(a));
if (getsockname(listener, &a.addr, &addrlen) == SOCKET_ERROR)
break;
// win32 getsockname may only set the port number, p=0.0005.
// ( http://msdn.microsoft.com/library/ms738543.aspx ):
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
if (listen(listener, 1) == SOCKET_ERROR)
break;
socks[0] = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
if (socks[0] == INVALID_SOCKET)
break;
if (connect(socks[0], &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
break;
socks[1] = accept(listener, NULL, NULL);
if (socks[1] == INVALID_SOCKET)
break;
closesocket(listener);
return 0;
}
e = WSAGetLastError();
closesocket(listener);
closesocket(socks[0]);
closesocket(socks[1]);
WSASetLastError(e);
socks[0] = socks[1] = -1;
return SOCKET_ERROR;
}
#else
int dumb_socketpair(int socks[2], int dummy)
{
if (socks == 0) {
errno = EINVAL;
return -1;
}
dummy = socketpair(AF_LOCAL, SOCK_STREAM, 0, socks);
if (dummy)
socks[0] = socks[1] = -1;
return dummy;
}
#endif
}}
\ No newline at end of file
......@@ -112,7 +112,8 @@ void publish_impl(abstract_actor_ptr ptr, std::unique_ptr<acceptor> aptr) {
abstract_actor_ptr remote_actor_impl(stream_ptr_pair io, string_set expected) {
CPPA_LOGF_TRACE("io{" << io.first.get() << ", " << io.second.get() << "}");
auto pinf = node_id::get();
auto mm = get_middleman();
auto pinf = mm->node();
std::uint32_t process_id = pinf->process_id();
// throws on error
io.second->write(&process_id, sizeof(std::uint32_t));
......@@ -196,7 +197,6 @@ abstract_actor_ptr remote_actor_impl(stream_ptr_pair io, string_set expected) {
auto ptr = get_actor_registry()->get(remote_aid);
return ptr;
}
auto mm = get_middleman();
struct remote_actor_result { remote_actor_result* next; actor value; };
intrusive::blocking_single_reader_queue<remote_actor_result> q;
mm->run_later([mm, io, pinfptr, remote_aid, &q] {
......
......@@ -171,7 +171,7 @@ int main() {
actor_namespace addressing;
cout << "process id: " << to_string(node_id::get()) << endl;
cout << "process id: " << to_string(get_middleman()->node()) << endl;
auto oarr = new detail::object_array;
oarr->push_back(object::from(static_cast<uint32_t>(42)));
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment