Commit a756eee3 authored by Richard Newton's avatar Richard Newton

Merge branch 'master' of https://github.com/zeromq/libzmq

parents ca94cf56 6b4d9bca
......@@ -101,6 +101,10 @@ test_xpub_nodrop
test_xpub_manual
test_xpub_welcome_msg
test_atomics
test_client_drop_more
test_client_server
test_server_drop_more
test_thread_safe
tests/test*.log
tests/test*.trs
src/platform.hpp*
......
......@@ -158,10 +158,6 @@ set(CMAKE_REQUIRED_INCLUDES )
add_definitions(-D_REENTRANT -D_THREAD_SAFE)
if(WIN32)
add_definitions(-DDLL_EXPORT)
endif()
option(ENABLE_EVENTFD "Enable/disable eventfd" ZMQ_HAVE_EVENTFD)
macro(zmq_check_cxx_flag_prepend flag)
......@@ -342,14 +338,13 @@ set(LIBRARY_OUTPUT_PATH ${CMAKE_CURRENT_BINARY_DIR}/lib)
#-----------------------------------------------------------------------------
# platform specifics
if(MSVC)
add_definitions(
-DWIN32
-DDLL_EXPORT
# NB: May require tweaking for highly connected applications.
-DFD_SETSIZE=4096
-D_CRT_SECURE_NO_WARNINGS)
if (WIN32)
# NB: May require tweaking for highly connected applications.
add_definitions (-DFD_SETSIZE=4096)
add_definitions (-D_CRT_SECURE_NO_WARNINGS)
endif ()
if(MSVC)
# Parallel make.
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /MP")
......@@ -387,6 +382,7 @@ set(cxx-sources
kqueue.cpp
lb.cpp
mailbox.cpp
mailbox_safe.cpp
mechanism.cpp
metadata.cpp
msg.cpp
......@@ -584,7 +580,8 @@ if(MSVC)
PUBLIC_HEADER "${public_headers}"
RELEASE_POSTFIX "${_zmq_COMPILER}-mt-${ZMQ_VERSION_MAJOR}_${ZMQ_VERSION_MINOR}_${ZMQ_VERSION_PATCH}"
DEBUG_POSTFIX "${_zmq_COMPILER}-mt-gd-${ZMQ_VERSION_MAJOR}_${ZMQ_VERSION_MINOR}_${ZMQ_VERSION_PATCH}"
RUNTIME_OUTPUT_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/bin")
RUNTIME_OUTPUT_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/bin"
COMPILE_DEFINITIONS "DLL_EXPORT")
add_library(libzmq-static STATIC ${sources})
set_target_properties(libzmq-static PROPERTIES
PUBLIC_HEADER "${public_headers}"
......@@ -594,16 +591,18 @@ if(MSVC)
OUTPUT_NAME "libzmq")
else()
add_library(libzmq SHARED ${sources} ${public_headers} ${html-docs} ${readme-docs} ${zmq-pkgconfig})
set_target_properties(libzmq PROPERTIES
COMPILE_DEFINITIONS "DLL_EXPORT"
PUBLIC_HEADER "${public_headers}"
VERSION ${ZMQ_VERSION}
SOVERSION "${ZMQ_VERSION_MAJOR}.${ZMQ_VERSION_MINOR}.0")
if(ZMQ_BUILD_FRAMEWORK)
set_target_properties(libzmq PROPERTIES
FRAMEWORK TRUE
OUTPUT_NAME "ZeroMQ"
PUBLIC_HEADER "${public_headers}"
MACOSX_FRAMEWORK_IDENTIFIER "org.zeromq.libzmq"
MACOSX_FRAMEWORK_SHORT_VERSION_STRING ${ZMQ_VERSION}
MACOSX_FRAMEWORK_BUNDLE_VERSION ${ZMQ_VERSION}
VERSION ${ZMQ_VERSION}
SOVERSION "${ZMQ_VERSION_MAJOR}.${ZMQ_VERSION_MINOR}.0")
MACOSX_FRAMEWORK_BUNDLE_VERSION ${ZMQ_VERSION})
set_source_files_properties(${html-docs} PROPERTIES
MACOSX_PACKAGE_LOCATION doc)
set_source_files_properties(${readme-docs} PROPERTIES
......@@ -612,13 +611,13 @@ else()
MACOSX_PACKAGE_LOCATION lib/pkgconfig)
else()
set_target_properties(libzmq PROPERTIES
OUTPUT_NAME "zmq"
PUBLIC_HEADER "${public_headers}")
OUTPUT_NAME "zmq"
)
endif()
add_library(libzmq-static STATIC ${sources} ${public_headers} ${html-docs} ${readme-docs} ${zmq-pkgconfig})
set_target_properties(libzmq-static PROPERTIES
PUBLIC_HEADER "${public_headers}"
COMPILE_FLAGS "-DZMQ_STATIC"
COMPILE_DEFINITIONS "ZMQ_STATIC"
OUTPUT_NAME "zmq-static")
endif()
......@@ -720,11 +719,11 @@ endif()
# DESTINATION include
# COMPONENT SDK)
if(NOT ZMQ_BUILD_FRAMEWORK)
file(GLOB private_headers "${CMAKE_CURRENT_SOURCE_DIR}/src/*.hpp")
install(FILES ${sources} ${private_headers} DESTINATION src/zmq
COMPONENT SourceCode)
endif()
#if(NOT ZMQ_BUILD_FRAMEWORK)
# file(GLOB private_headers "${CMAKE_CURRENT_SOURCE_DIR}/src/*.hpp")
# install(FILES ${sources} ${private_headers} DESTINATION src/zmq
# COMPONENT SourceCode)
#endif()
foreach(readme ${readme-docs})
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/${readme} ${CMAKE_CURRENT_BINARY_DIR}/${readme}.txt)
......@@ -733,7 +732,7 @@ foreach(readme ${readme-docs})
if(MSVC)
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/${readme}.txt DESTINATION .)
else()
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/${readme}.txt DESTINATION etc/zmq)
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/${readme}.txt DESTINATION share/zmq)
endif()
endif()
endforeach()
......
......@@ -31,6 +31,7 @@ src_libzmq_la_SOURCES = \
src/clock.cpp \
src/clock.hpp \
src/command.hpp \
src/condition_variable.hpp \
src/config.hpp \
src/ctx.cpp \
src/ctx.hpp \
......@@ -63,6 +64,7 @@ src_libzmq_la_SOURCES = \
src/i_encoder.hpp \
src/i_engine.hpp \
src/i_decoder.hpp \
src/i_mailbox.hpp \
src/i_poll_events.hpp \
src/io_object.cpp \
src/io_object.hpp \
......@@ -83,6 +85,8 @@ src_libzmq_la_SOURCES = \
src/likely.hpp \
src/mailbox.cpp \
src/mailbox.hpp \
src/mailbox_safe.cpp \
src/mailbox_safe.hpp \
src/mechanism.cpp \
src/mechanism.hpp \
src/metadata.cpp \
......@@ -254,6 +258,7 @@ src_libzmq_la_CPPFLAGS += ${pgm_CFLAGS}
src_libzmq_la_LIBADD += ${pgm_LIBS}
endif
if ENABLE_PERF
noinst_PROGRAMS = \
perf/local_lat \
perf/remote_lat \
......@@ -279,11 +284,14 @@ perf_inproc_lat_SOURCES = perf/inproc_lat.cpp
perf_inproc_thr_LDADD = src/libzmq.la
perf_inproc_thr_SOURCES = perf/inproc_thr.cpp
endif
if ENABLE_CURVE_KEYGEN
bin_PROGRAMS = tools/curve_keygen
tools_curve_keygen_LDADD = src/libzmq.la
tools_curve_keygen_SOURCES = tools/curve_keygen.cpp
endif
#
# tests
......@@ -345,7 +353,8 @@ test_apps = \
tests/test_atomics \
tests/test_client_server \
tests/test_server_drop_more \
tests/test_client_drop_more
tests/test_client_drop_more \
tests/test_thread_safe
tests_test_system_SOURCES = tests/test_system.cpp
tests_test_system_LDADD = src/libzmq.la
......@@ -526,6 +535,11 @@ tests_test_server_drop_more_LDADD = src/libzmq.la
tests_test_client_drop_more_SOURCES = tests/test_client_drop_more.cpp
tests_test_client_drop_more_LDADD = src/libzmq.la
tests_test_thread_safe_SOURCES = tests/test_thread_safe.cpp
tests_test_thread_safe_LDADD = src/libzmq.la
if !ON_MINGW
if !ON_CYGWIN
test_apps += \
......
......@@ -586,8 +586,8 @@ dnl # LIBZMQ_CHECK_SOCK_CLOEXEC([action-if-found], [action-if-not-found])
dnl # Check if SOCK_CLOEXEC is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_SOCK_CLOEXEC], [{
AC_MSG_CHECKING(whether SOCK_CLOEXEC is supported)
AC_TRY_RUN([/* SOCK_CLOEXEC test */
AC_CACHE_CHECK([whether SOCK_CLOEXEC is supported], [libzmq_cv_sock_cloexec],
[AC_TRY_RUN([/* SOCK_CLOEXEC test */
#include <sys/types.h>
#include <sys/socket.h>
......@@ -596,11 +596,13 @@ int main (int argc, char *argv [])
int s = socket (PF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
return (s == -1);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_sock_cloexec="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_sock_cloexec="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_sock_cloexec="no"]
],
[libzmq_cv_sock_cloexec="yes"],
[libzmq_cv_sock_cloexec="no"],
[libzmq_cv_sock_cloexec="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_sock_cloexec" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......@@ -628,8 +630,8 @@ dnl # LIBZMQ_CHECK_SO_KEEPALIVE([action-if-found], [action-if-not-found])
dnl # Check if SO_KEEPALIVE is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_SO_KEEPALIVE], [{
AC_MSG_CHECKING(whether SO_KEEPALIVE is supported)
AC_TRY_RUN([/* SO_KEEPALIVE test */
AC_CACHE_CHECK([whether SO_KEEPALIVE is supported], [libzmq_cv_so_keepalive],
[AC_TRY_RUN([/* SO_KEEPALIVE test */
#include <sys/types.h>
#include <sys/socket.h>
......@@ -641,11 +643,13 @@ int main (int argc, char *argv [])
((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_so_keepalive="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_so_keepalive="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_so_keepalive="no"]
],
[libzmq_cv_so_keepalive="yes"],
[libzmq_cv_so_keepalive="no"],
[libzmq_cv_so_keepalive="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_so_keepalive" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......@@ -653,8 +657,8 @@ dnl # LIBZMQ_CHECK_TCP_KEEPCNT([action-if-found], [action-if-not-found])
dnl # Check if TCP_KEEPCNT is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPCNT], [{
AC_MSG_CHECKING(whether TCP_KEEPCNT is supported)
AC_TRY_RUN([/* TCP_KEEPCNT test */
AC_CACHE_CHECK([whether TCP_KEEPCNT is supported], [libzmq_cv_tcp_keepcnt],
[AC_TRY_RUN([/* TCP_KEEPCNT test */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
......@@ -669,11 +673,13 @@ int main (int argc, char *argv [])
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPCNT, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepcnt="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_tcp_keepcnt="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepcnt="no"]
],
[libzmq_cv_tcp_keepcnt="yes"],
[libzmq_cv_tcp_keepcnt="no"],
[libzmq_cv_tcp_keepcnt="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_tcp_keepcnt" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......@@ -681,8 +687,8 @@ dnl # LIBZMQ_CHECK_TCP_KEEPIDLE([action-if-found], [action-if-not-found])
dnl # Check if TCP_KEEPIDLE is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPIDLE], [{
AC_MSG_CHECKING(whether TCP_KEEPIDLE is supported)
AC_TRY_RUN([/* TCP_KEEPIDLE test */
AC_CACHE_CHECK([whether TCP_KEEPIDLE is supported], [libzmq_cv_tcp_keepidle],
[AC_TRY_RUN([/* TCP_KEEPIDLE test */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
......@@ -697,11 +703,13 @@ int main (int argc, char *argv [])
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPIDLE, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepidle="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_tcp_keepidle="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepidle="no"]
],
[libzmq_cv_tcp_keepidle="yes"],
[libzmq_cv_tcp_keepidle="no"],
[libzmq_cv_tcp_keepidle="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_tcp_keepidle" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......@@ -709,8 +717,8 @@ dnl # LIBZMQ_CHECK_TCP_KEEPINTVL([action-if-found], [action-if-not-found])
dnl # Check if TCP_KEEPINTVL is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPINTVL], [{
AC_MSG_CHECKING(whether TCP_KEEPINTVL is supported)
AC_TRY_RUN([/* TCP_KEEPINTVL test */
AC_CACHE_CHECK([whether TCP_KEEPINTVL is supported], [libzmq_cv_tcp_keepintvl],
[AC_TRY_RUN([/* TCP_KEEPINTVL test */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
......@@ -725,11 +733,13 @@ int main (int argc, char *argv [])
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPINTVL, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepintvl="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_tcp_keepintvl="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepintvl="no"]
],
[libzmq_cv_tcp_keepintvl="yes"],
[libzmq_cv_tcp_keepintvl="no"],
[libzmq_cv_tcp_keepintvl="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_tcp_keepintvl" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......@@ -737,8 +747,8 @@ dnl # LIBZMQ_CHECK_TCP_KEEPALIVE([action-if-found], [action-if-not-found])
dnl # Check if TCP_KEEPALIVE is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPALIVE], [{
AC_MSG_CHECKING(whether TCP_KEEPALIVE is supported)
AC_TRY_RUN([/* TCP_KEEPALIVE test */
AC_CACHE_CHECK([whether TCP_KEEPALIVE is supported], [libzmq_cv_tcp_keepalive],
[AC_TRY_RUN([/* TCP_KEEPALIVE test */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
......@@ -753,11 +763,13 @@ int main (int argc, char *argv [])
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPALIVE, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepalive="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_tcp_keepalive="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepalive="no"]
],
[libzmq_cv_tcp_keepalive="yes"],
[libzmq_cv_tcp_keepalive="no"],
[libzmq_cv_tcp_keepalive="not during cross-compile"]
)]
)
AS_IF([test "x$libzmq_cv_tcp_keepalive" = "xyes"], [$1], [$2])
}])
dnl ################################################################################
......
......@@ -6,19 +6,21 @@
# A2X_EXECUTABLE - the full path to a2x
# A2X_FOUND - If false, don't attempt to use a2x.
set (PROGRAMFILESX86 "PROGRAMFILES(X86)")
find_program(ASCIIDOC_EXECUTABLE asciidoc asciidoc.py
PATHS "$ENV{ASCIIDOC_ROOT}"
"$ENV{PROGRAMW6432}/asciidoc"
"$ENV{PROGRAMFILES}/asciidoc"
"$ENV{PROGRAMFILES(X86)}/asciidoc")
"$ENV{${PROGRAMFILESX86}}/asciidoc")
find_program(A2X_EXECUTABLE a2x
PATHS "$ENV{ASCIIDOC_ROOT}"
"$ENV{PROGRAMW6432}/asciidoc"
"$ENV{PROGRAMFILES}/asciidoc"
"$ENV{PROGRAMFILES(X86)}/asciidoc")
"$ENV{${PROGRAMFILESX86}}/asciidoc")
include(FindPackageHandleStandardArgs)
find_package_handle_standard_ARGS(AsciiDoc REQUIRED_VARS ASCIIDOC_EXECUTABLE)
mark_as_advanced(ASCIIDOC_EXECUTABLE A2X_EXECUTABLE)
\ No newline at end of file
mark_as_advanced(ASCIIDOC_EXECUTABLE A2X_EXECUTABLE)
......@@ -86,5 +86,14 @@
#cmakedefine ZMQ_HAVE_WINDOWS
#ifdef ZMQ_HAVE_WINDOWS
#if defined _WIN32_WINNT && _WIN32_WINNT < 0x0600
#undef _WIN32_WINNT
#endif
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0600
#endif
#endif
#endif
\ No newline at end of file
......@@ -29,4 +29,13 @@
#define ZMQ_HAVE_WINDOWS
#if defined _WIN32_WINNT && _WIN32_WINNT < 0x0600
#undef _WIN32_WINNT
#endif
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0600
#endif
#endif
......@@ -29,4 +29,12 @@
#define ZMQ_HAVE_WINDOWS
#if defined _WIN32_WINNT && _WIN32_WINNT < 0x0600
#undef _WIN32_WINNT
#endif
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0600
#endif
#endif
......@@ -13,7 +13,7 @@
<ItemDefinitionGroup>
<ClCompile>
<RuntimeLibrary>MultiThreadedDebugDLL</RuntimeLibrary>
<RuntimeLibrary>MultiThreadedDLL</RuntimeLibrary>
</ClCompile>
</ItemDefinitionGroup>
......
B// Microsoft Visual C++ generated resource script.
......
......@@ -73,8 +73,10 @@
<ClInclude Include="..\..\..\..\src\array.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_counter.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_ptr.hpp" />
<ClInclude Include="..\..\..\..\src\client.hpp" />
<ClInclude Include="..\..\..\..\src\clock.hpp" />
<ClInclude Include="..\..\..\..\src\command.hpp" />
<ClInclude Include="..\..\..\..\src\condition_variable.hpp" />
<ClInclude Include="..\..\..\..\src\config.hpp" />
<ClInclude Include="..\..\..\..\src\ctx.hpp" />
<ClInclude Include="..\..\..\..\src\curve_client.hpp" />
......@@ -102,6 +104,7 @@
<ClInclude Include="..\..\..\..\src\lb.hpp" />
<ClInclude Include="..\..\..\..\src\likely.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp" />
<ClInclude Include="..\..\..\..\src\msg.hpp" />
<ClInclude Include="..\..\..\..\src\mtrie.hpp" />
<ClInclude Include="..\..\..\..\src\mutex.hpp" />
......@@ -113,6 +116,7 @@
<ClInclude Include="..\..\..\..\src\pgm_sender.hpp" />
<ClInclude Include="..\..\..\..\src\pgm_socket.hpp" />
<ClInclude Include="..\..\..\..\src\pipe.hpp" />
<ClInclude Include="..\..\..\..\src\server.hpp" />
<ClInclude Include="..\..\..\..\src\socks.hpp" />
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp" />
<ClInclude Include="..\..\platform.hpp" />
......@@ -158,6 +162,7 @@
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\..\..\src\address.cpp" />
<ClCompile Include="..\..\..\..\src\client.cpp" />
<ClCompile Include="..\..\..\..\src\clock.cpp" />
<ClCompile Include="..\..\..\..\src\ctx.cpp" />
<ClCompile Include="..\..\..\..\src\curve_client.cpp" />
......@@ -180,6 +185,7 @@
<ClCompile Include="..\..\..\..\src\kqueue.cpp" />
<ClCompile Include="..\..\..\..\src\lb.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp" />
<ClCompile Include="..\..\..\..\src\mechanism.cpp" />
<ClCompile Include="..\..\..\..\src\metadata.cpp" />
<ClCompile Include="..\..\..\..\src\msg.cpp" />
......@@ -210,6 +216,7 @@
<ClCompile Include="..\..\..\..\src\req.cpp" />
<ClCompile Include="..\..\..\..\src\router.cpp" />
<ClCompile Include="..\..\..\..\src\select.cpp" />
<ClCompile Include="..\..\..\..\src\server.cpp" />
<ClCompile Include="..\..\..\..\src\session_base.cpp" />
<ClCompile Include="..\..\..\..\src\signaler.cpp" />
<ClCompile Include="..\..\..\..\src\socket_base.cpp" />
......@@ -247,4 +254,4 @@
<ResourceCompile Include="..\..\resource.rc" />
</ItemGroup>
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
</Project>
</Project>
\ No newline at end of file
......@@ -226,6 +226,15 @@
<ClCompile Include="..\..\..\..\src\socks_connecter.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\server.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\client.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp">
<Filter>src</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\..\..\include\zmq_utils.h">
......@@ -492,6 +501,18 @@
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\server.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\client.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\condition_variable.hpp">
<Filter>src\include</Filter>
</ClInclude>
</ItemGroup>
<ItemGroup>
<Filter Include="include">
......
......@@ -73,8 +73,10 @@
<ClInclude Include="..\..\..\..\src\array.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_counter.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_ptr.hpp" />
<ClInclude Include="..\..\..\..\src\client.hpp" />
<ClInclude Include="..\..\..\..\src\clock.hpp" />
<ClInclude Include="..\..\..\..\src\command.hpp" />
<ClInclude Include="..\..\..\..\src\condition_variable.hpp" />
<ClInclude Include="..\..\..\..\src\config.hpp" />
<ClInclude Include="..\..\..\..\src\ctx.hpp" />
<ClInclude Include="..\..\..\..\src\curve_client.hpp" />
......@@ -102,6 +104,7 @@
<ClInclude Include="..\..\..\..\src\lb.hpp" />
<ClInclude Include="..\..\..\..\src\likely.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp" />
<ClInclude Include="..\..\..\..\src\msg.hpp" />
<ClInclude Include="..\..\..\..\src\mtrie.hpp" />
<ClInclude Include="..\..\..\..\src\mutex.hpp" />
......@@ -113,6 +116,7 @@
<ClInclude Include="..\..\..\..\src\pgm_sender.hpp" />
<ClInclude Include="..\..\..\..\src\pgm_socket.hpp" />
<ClInclude Include="..\..\..\..\src\pipe.hpp" />
<ClInclude Include="..\..\..\..\src\server.hpp" />
<ClInclude Include="..\..\..\..\src\socks.hpp" />
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp" />
<ClInclude Include="..\..\platform.hpp" />
......@@ -158,6 +162,7 @@
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\..\..\src\address.cpp" />
<ClCompile Include="..\..\..\..\src\client.cpp" />
<ClCompile Include="..\..\..\..\src\clock.cpp" />
<ClCompile Include="..\..\..\..\src\ctx.cpp" />
<ClCompile Include="..\..\..\..\src\curve_client.cpp" />
......@@ -180,6 +185,7 @@
<ClCompile Include="..\..\..\..\src\kqueue.cpp" />
<ClCompile Include="..\..\..\..\src\lb.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp" />
<ClCompile Include="..\..\..\..\src\mechanism.cpp" />
<ClCompile Include="..\..\..\..\src\metadata.cpp" />
<ClCompile Include="..\..\..\..\src\msg.cpp" />
......@@ -210,6 +216,7 @@
<ClCompile Include="..\..\..\..\src\req.cpp" />
<ClCompile Include="..\..\..\..\src\router.cpp" />
<ClCompile Include="..\..\..\..\src\select.cpp" />
<ClCompile Include="..\..\..\..\src\server.cpp" />
<ClCompile Include="..\..\..\..\src\session_base.cpp" />
<ClCompile Include="..\..\..\..\src\signaler.cpp" />
<ClCompile Include="..\..\..\..\src\socket_base.cpp" />
......@@ -247,4 +254,4 @@
<ResourceCompile Include="..\..\resource.rc" />
</ItemGroup>
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
</Project>
</Project>
\ No newline at end of file
......@@ -226,6 +226,15 @@
<ClCompile Include="..\..\..\..\src\socks_connecter.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\server.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\client.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp">
<Filter>src</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\..\..\include\zmq_utils.h">
......@@ -492,6 +501,18 @@
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\server.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\client.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\condition_variable.hpp">
<Filter>src\include</Filter>
</ClInclude>
</ItemGroup>
<ItemGroup>
<Filter Include="include">
......
......@@ -73,8 +73,10 @@
<ClInclude Include="..\..\..\..\src\array.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_counter.hpp" />
<ClInclude Include="..\..\..\..\src\atomic_ptr.hpp" />
<ClInclude Include="..\..\..\..\src\client.hpp" />
<ClInclude Include="..\..\..\..\src\clock.hpp" />
<ClInclude Include="..\..\..\..\src\command.hpp" />
<ClInclude Include="..\..\..\..\src\condition_variable.hpp" />
<ClInclude Include="..\..\..\..\src\config.hpp" />
<ClInclude Include="..\..\..\..\src\ctx.hpp" />
<ClInclude Include="..\..\..\..\src\curve_client.hpp" />
......@@ -102,6 +104,7 @@
<ClInclude Include="..\..\..\..\src\lb.hpp" />
<ClInclude Include="..\..\..\..\src\likely.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox.hpp" />
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp" />
<ClInclude Include="..\..\..\..\src\msg.hpp" />
<ClInclude Include="..\..\..\..\src\mtrie.hpp" />
<ClInclude Include="..\..\..\..\src\mutex.hpp" />
......@@ -113,6 +116,7 @@
<ClInclude Include="..\..\..\..\src\pgm_sender.hpp" />
<ClInclude Include="..\..\..\..\src\pgm_socket.hpp" />
<ClInclude Include="..\..\..\..\src\pipe.hpp" />
<ClInclude Include="..\..\..\..\src\server.hpp" />
<ClInclude Include="..\..\..\..\src\socks.hpp" />
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp" />
<ClInclude Include="..\..\platform.hpp" />
......@@ -158,6 +162,7 @@
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\..\..\src\address.cpp" />
<ClCompile Include="..\..\..\..\src\client.cpp" />
<ClCompile Include="..\..\..\..\src\clock.cpp" />
<ClCompile Include="..\..\..\..\src\ctx.cpp" />
<ClCompile Include="..\..\..\..\src\curve_client.cpp" />
......@@ -180,6 +185,7 @@
<ClCompile Include="..\..\..\..\src\kqueue.cpp" />
<ClCompile Include="..\..\..\..\src\lb.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox.cpp" />
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp" />
<ClCompile Include="..\..\..\..\src\mechanism.cpp" />
<ClCompile Include="..\..\..\..\src\metadata.cpp" />
<ClCompile Include="..\..\..\..\src\msg.cpp" />
......@@ -210,6 +216,7 @@
<ClCompile Include="..\..\..\..\src\req.cpp" />
<ClCompile Include="..\..\..\..\src\router.cpp" />
<ClCompile Include="..\..\..\..\src\select.cpp" />
<ClCompile Include="..\..\..\..\src\server.cpp" />
<ClCompile Include="..\..\..\..\src\session_base.cpp" />
<ClCompile Include="..\..\..\..\src\signaler.cpp" />
<ClCompile Include="..\..\..\..\src\socket_base.cpp" />
......@@ -247,4 +254,4 @@
<ResourceCompile Include="..\..\resource.rc" />
</ItemGroup>
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
</Project>
</Project>
\ No newline at end of file
......@@ -226,6 +226,15 @@
<ClCompile Include="..\..\..\..\src\socks_connecter.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\server.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\client.cpp">
<Filter>src</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\src\mailbox_safe.cpp">
<Filter>src</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\..\..\include\zmq_utils.h">
......@@ -492,6 +501,18 @@
<ClInclude Include="..\..\..\..\src\socks_connecter.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\server.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\client.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\mailbox_safe.hpp">
<Filter>src\include</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\src\condition_variable.hpp">
<Filter>src\include</Filter>
</ClInclude>
</ItemGroup>
<ItemGroup>
<Filter Include="include">
......
......@@ -346,6 +346,22 @@ if test "x$zmq_enable_eventfd" = "xyes"; then
[AC_DEFINE(ZMQ_HAVE_EVENTFD, 1, [Have eventfd extension.])])
fi
# Conditionally build performance measurement tools
AC_ARG_ENABLE([perf],
[AS_HELP_STRING([--enable-perf], [Build performance measurement tools [default=yes].])],
[zmq_enable_perf=$enableval],
[zmq_enable_perf=yes])
AM_CONDITIONAL(ENABLE_PERF, test "x$zmq_enable_perf" = "xyes")
# Conditionally build curve key generation tool
AC_ARG_ENABLE([curve-keygen],
[AS_HELP_STRING([--enable-curve-keygen], [Build curve key-generation tool [default=yes].])],
[zmq_enable_curve_keygen=$enableval],
[zmq_enable_curve_keygen=yes])
AM_CONDITIONAL(ENABLE_CURVE_KEYGEN, test "x$zmq_enable_curve_keygen" = "xyes")
# Use c++ in subsequent tests
AC_LANG_PUSH(C++)
......
......@@ -4,7 +4,7 @@ zmq_ctx_term(3)
NAME
----
zmq_ctx_term - destroy a 0MQ context
zmq_ctx_term - terminate a 0MQ context
SYNOPSIS
......@@ -36,7 +36,8 @@ Context termination is performed in the following steps:
For further details regarding socket linger behaviour refer to the _ZMQ_LINGER_
option in linkzmq:zmq_setsockopt[3].
This function replaces the deprecated function linkzmq:zmq_term[3].
This function replaces the deprecated functions linkzmq:zmq_term[3] and
linkzmq:zmq_ctx_destroy[3].
RETURN VALUE
......
......@@ -28,8 +28,7 @@ _zmq_msg_init_size()_ are mutually exclusive. Never initialise the same
RETURN VALUE
------------
The _zmq_msg_init()_ function shall return zero if successful. Otherwise it
shall return `-1` and set 'errno' to one of the values defined below.
The _zmq_msg_init()_ function always returns zero.
ERRORS
......
......@@ -40,6 +40,8 @@ The endpoint supplied is invalid.
The 0MQ 'context' associated with the specified 'socket' was terminated.
*ENOTSOCK*::
The provided 'socket' was invalid.
*ENOENT*::
The endpoint supplied was not previously bound.
EXAMPLES
......
<?xml version="1.0" encoding="utf-8"?>
<!-- These values are populated into the package.gsl templates by package.bat. -->
<!-- The target attribute controls path and file name only, id controls package naming. -->
<package id="libzmq_vc120" target="libzmq" version = "4.1.0.7" pathversion="4_1_0_7" platformtoolset="v120">
<package id="libzmq_vc120" target="libzmq" version = "4.2.0.0" pathversion="4_2_0_0" platformtoolset="v120">
<dependency id="libsodium_vc120" version="0.5.0.0" />
</package>
\ No newline at end of file
......@@ -7,7 +7,7 @@
<package xmlns="http://schemas.microsoft.com/packaging/2013/01/nuspec.xsd">
<metadata minClientVersion="2.5">
<id>libzmq_vc120</id>
<version>4.1.0.7</version>
<version>4.2.0.0</version>
<title>libzmq_vc120</title>
<authors>libzmq contributors</authors>
<owners>Eric Voskuil</owners>
......@@ -44,52 +44,52 @@
<!-- libraries -->
<!-- x86 Dynamic libraries (.dll) -->
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x86-v120-mt-4_1_0_7.dll" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x86-v120-mt-gd-4_1_0_7.dll" />
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x86-v120-mt-4_2_0_0.dll" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x86-v120-mt-gd-4_2_0_0.dll" />
<!-- x86 Debugging symbols (.pdb) -->
<!--<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x86-v120-mt-4_1_0_7.pdb" />-->
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x86-v120-mt-gd-4_1_0_7.pdb" />
<!--<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x86-v120-mt-4_2_0_0.pdb" />-->
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x86-v120-mt-gd-4_2_0_0.pdb" />
<!-- x86 Import libraries (.imp.lib) -->
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-4_1_0_7.imp.lib" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-gd-4_1_0_7.imp.lib" />
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-4_2_0_0.imp.lib" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-gd-4_2_0_0.imp.lib" />
<!-- x86 Export libraries (.exp) -->
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x86-v120-mt-4_1_0_7.exp" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x86-v120-mt-gd-4_1_0_7.exp" />
<file src="..\..\bin\Win32\Release\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x86-v120-mt-4_2_0_0.exp" />
<file src="..\..\bin\Win32\Debug\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x86-v120-mt-gd-4_2_0_0.exp" />
<!-- x86 Static libraries (.lib) -->
<file src="..\..\bin\Win32\Release\v120\static\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-s-4_1_0_7.lib" />
<file src="..\..\bin\Win32\Debug\v120\static\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-sgd-4_1_0_7.lib" />
<file src="..\..\bin\Win32\Release\v120\static\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-s-4_2_0_0.lib" />
<file src="..\..\bin\Win32\Debug\v120\static\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-sgd-4_2_0_0.lib" />
<!-- x86 Static link time code generation libraries (.ltcg.lib) -->
<file src="..\..\bin\Win32\Release\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-s-4_1_0_7.ltcg.lib" />
<file src="..\..\bin\Win32\Debug\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-sgd-4_1_0_7.ltcg.lib" />
<file src="..\..\bin\Win32\Release\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-s-4_2_0_0.ltcg.lib" />
<file src="..\..\bin\Win32\Debug\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x86-v120-mt-sgd-4_2_0_0.ltcg.lib" />
<!-- x64 Dynamic libraries (.dll) -->
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x64-v120-mt-4_1_0_7.dll" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x64-v120-mt-gd-4_1_0_7.dll" />
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x64-v120-mt-4_2_0_0.dll" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.dll" target="build\native\bin\libzmq-x64-v120-mt-gd-4_2_0_0.dll" />
<!-- x64 Debugging symbols (.pdb) -->
<!--<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x64-v120-mt-4_1_0_7.pdb" />-->
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x64-v120-mt-gd-4_1_0_7.pdb" />
<!--<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x64-v120-mt-4_2_0_0.pdb" />-->
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.pdb" target="build\native\bin\libzmq-x64-v120-mt-gd-4_2_0_0.pdb" />
<!-- x64 Import libraries (.imp.lib) -->
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-4_1_0_7.imp.lib" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-gd-4_1_0_7.imp.lib" />
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-4_2_0_0.imp.lib" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-gd-4_2_0_0.imp.lib" />
<!-- x64 Export libraries (.exp) -->
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x64-v120-mt-4_1_0_7.exp" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x64-v120-mt-gd-4_1_0_7.exp" />
<file src="..\..\bin\x64\Release\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x64-v120-mt-4_2_0_0.exp" />
<file src="..\..\bin\x64\Debug\v120\dynamic\libzmq.exp" target="build\native\bin\libzmq-x64-v120-mt-gd-4_2_0_0.exp" />
<!-- x64 Static libraries (.lib) -->
<file src="..\..\bin\x64\Release\v120\static\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-s-4_1_0_7.lib" />
<file src="..\..\bin\x64\Debug\v120\static\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-sgd-4_1_0_7.lib" />
<file src="..\..\bin\x64\Release\v120\static\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-s-4_2_0_0.lib" />
<file src="..\..\bin\x64\Debug\v120\static\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-sgd-4_2_0_0.lib" />
<!-- x64 Static link time code generation libraries (.ltcg.lib) -->
<file src="..\..\bin\Win32\Release\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-s-4_1_0_7.ltcg.lib" />
<file src="..\..\bin\Win32\Debug\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-sgd-4_1_0_7.ltcg.lib" />
<file src="..\..\bin\Win32\Release\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-s-4_2_0_0.ltcg.lib" />
<file src="..\..\bin\Win32\Debug\v120\ltcg\libzmq.lib" target="build\native\bin\libzmq-x64-v120-mt-sgd-4_2_0_0.ltcg.lib" />
</files>
<!--
#################################################################
......
......@@ -32,66 +32,66 @@
<!-- static libraries -->
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'static' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-s-4_1_0_7.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-s-4_2_0_0.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'static' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-sgd-4_1_0_7.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-sgd-4_2_0_0.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'static' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-s-4_1_0_7.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-s-4_2_0_0.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'static' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-sgd-4_1_0_7.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-sgd-4_2_0_0.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<!-- static ltcg libraries -->
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'ltcg' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-s-4_1_0_7.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-s-4_2_0_0.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'ltcg' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-sgd-4_1_0_7.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-sgd-4_2_0_0.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'ltcg' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-s-4_1_0_7.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-s-4_2_0_0.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'ltcg' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-sgd-4_1_0_7.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-sgd-4_2_0_0.ltcg.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<!-- dynamic import libraries -->
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-4_1_0_7.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-4_2_0_0.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x86-v120-mt-gd-4_1_0_7.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x86-v120-mt-gd-4_2_0_0.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Release')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-4_1_0_7.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-4_2_0_0.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Debug')) != -1">
<Link>
<AdditionalDependencies>libzmq-x64-v120-mt-gd-4_1_0_7.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalDependencies>libzmq-x64-v120-mt-gd-4_2_0_0.imp.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link>
</ItemDefinitionGroup>
......@@ -100,26 +100,26 @@
<Target Name="libzmq_AfterBuild_Win32_v120_Dynamic_Release"
Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Release')) != -1"
AfterTargets="libzmq_AfterBuild">
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-4_1_0_7.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<!--<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-4_1_0_7.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />-->
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-4_2_0_0.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<!--<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-4_2_0_0.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />-->
</Target>
<Target Name="libzmq_AfterBuild_Win32_v120_Dynamic_Debug"
Condition="'$(Platform)' == 'Win32' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Debug')) != -1"
AfterTargets="libzmq_AfterBuild">
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-gd-4_1_0_7.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-gd-4_1_0_7.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-gd-4_2_0_0.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x86-v120-mt-gd-4_2_0_0.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />
</Target>
<Target Name="libzmq_AfterBuild_x64_v120_Dynamic_Release"
Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Release')) != -1"
AfterTargets="libzmq_AfterBuild">
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-4_1_0_7.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<!--<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-4_1_0_7.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />-->
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-4_2_0_0.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<!--<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-4_2_0_0.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />-->
</Target>
<Target Name="libzmq_AfterBuild_x64_v120_Dynamic_Debug"
Condition="'$(Platform)' == 'x64' And ('$(PlatformToolset)' == 'v120' Or '$(PlatformToolset)' == 'CTP_Nov2013') And '$(Linkage-libzmq)' == 'dynamic' And $(Configuration.IndexOf('Debug')) != -1"
AfterTargets="libzmq_AfterBuild">
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-gd-4_1_0_7.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-gd-4_1_0_7.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-gd-4_2_0_0.dll" DestinationFiles="$(TargetDir)libzmq.dll" SkipUnchangedFiles="true" />
<Copy SourceFiles="$(MSBuildThisFileDirectory)bin\libzmq-x64-v120-mt-gd-4_2_0_0.pdb" DestinationFiles="$(TargetDir)libzmq.pdb" SkipUnchangedFiles="true" />
</Target>
<!--
......
......@@ -22,7 +22,7 @@
#include "msg.hpp"
zmq::client_t::client_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
socket_base_t (parent_, tid_, sid_)
socket_base_t (parent_, tid_, sid_, true)
{
options.type = ZMQ_CLIENT;
}
......@@ -99,4 +99,4 @@ void zmq::client_t::xpipe_terminated (pipe_t *pipe_)
{
fq.pipe_terminated (pipe_);
lb.pipe_terminated (pipe_);
}
\ No newline at end of file
}
......@@ -42,6 +42,31 @@
#include <time.h>
#endif
#if defined ZMQ_HAVE_OSX
#include <mach/clock.h>
#include <mach/mach.h>
#include <time.h>
#include <sys/time.h>
int clock_gettime (int clock_id, timespec *ts)
{
// The clock_id specified is not supported on this system.
if (clock_id != CLOCK_REALTIME) {
errno = EINVAL;
return -1;
}
clock_serv_t cclock;
mach_timespec_t mts;
host_get_clock_service (mach_host_self (), CALENDAR_CLOCK, &cclock);
clock_get_time (cclock, &mts);
mach_port_deallocate (mach_task_self (), cclock);
ts->tv_sec = mts.tv_sec;
ts->tv_nsec = mts.tv_nsec;
return 0;
}
#endif
#ifdef ZMQ_HAVE_WINDOWS
typedef ULONGLONG (*f_compatible_get_tick_count64)();
......
......@@ -22,6 +22,16 @@
#include "stdint.hpp"
#if defined ZMQ_HAVE_OSX
#include <mach/clock.h>
#include <mach/mach.h>
#include <time.h>
#include <sys/time.h>
int clock_gettime (int clock_id, timespec *ts);
#define CLOCK_REALTIME 0
#define HAVE_CLOCK_GETTIME
#endif
namespace zmq
{
......
/*
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ 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, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_CONDITON_VARIABLE_HPP_INCLUDED__
#define __ZMQ_CONDITON_VARIABLE_HPP_INCLUDED__
#include "platform.hpp"
#include "clock.hpp"
#include "err.hpp"
#include "mutex.hpp"
// Condition variable class encapsulates OS mutex in a platform-independent way.
#ifdef ZMQ_HAVE_WINDOWS
#include "windows.hpp"
namespace zmq
{
class condition_variable_t
{
public:
inline condition_variable_t ()
{
InitializeConditionVariable (&cv);
}
inline ~condition_variable_t ()
{
}
inline int wait (mutex_t* mutex_, int timeout_ )
{
int rc = SleepConditionVariableCS(&cv, mutex_->get_cs (), timeout_);
if (rc != 0)
return 0;
rc = GetLastError();
if (rc != ERROR_TIMEOUT)
win_assert(rc);
errno = EAGAIN;
return -1;
}
inline void broadcast ()
{
WakeAllConditionVariable(&cv);
}
private:
CONDITION_VARIABLE cv;
// Disable copy construction and assignment.
condition_variable_t (const condition_variable_t&);
void operator = (const condition_variable_t&);
};
}
#else
#include <pthread.h>
namespace zmq
{
class condition_variable_t
{
public:
inline condition_variable_t ()
{
int rc = pthread_cond_init (&cond, NULL);
posix_assert (rc);
}
inline ~condition_variable_t ()
{
int rc = pthread_cond_destroy (&cond);
posix_assert (rc);
}
inline int wait (mutex_t* mutex_, int timeout_)
{
int rc;
if (timeout_ != -1) {
struct timespec timeout;
clock_gettime(CLOCK_REALTIME, &timeout);
timeout.tv_sec += timeout_ / 1000;
timeout.tv_nsec += (timeout_ % 1000) * 1000000;
rc = pthread_cond_timedwait (&cond, mutex_->get_mutex (), &timeout);
}
else
rc = pthread_cond_wait(&cond, mutex_->get_mutex());
if (rc == 0)
return 0;
if (rc == ETIMEDOUT){
errno= EAGAIN;
return -1;
}
posix_assert (rc);
return -1;
}
inline void broadcast ()
{
int rc = pthread_cond_broadcast (&cond);
posix_assert (rc);
}
private:
pthread_cond_t cond;
// Disable copy construction and assignment.
condition_variable_t (const condition_variable_t&);
const condition_variable_t &operator = (const condition_variable_t&);
};
}
#endif
#endif
......@@ -273,7 +273,7 @@ zmq::socket_base_t *zmq::ctx_t::create_socket (int type_)
int ios = io_thread_count;
opt_sync.unlock ();
slot_count = mazmq + ios + 2;
slots = (mailbox_t **) malloc (sizeof (mailbox_t*) * slot_count);
slots = (i_mailbox **) malloc (sizeof (i_mailbox*) * slot_count);
alloc_assert (slots);
// Initialise the infrastructure for zmq_ctx_term thread.
......
......@@ -162,7 +162,7 @@ namespace zmq
// Array of pointers to mailboxes for both application and I/O threads.
uint32_t slot_count;
mailbox_t **slots;
i_mailbox **slots;
// Mailbox for zmq_term thread.
mailbox_t term_mailbox;
......
/*
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ 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, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_I_MAILBOX_HPP_INCLUDED__
#define __ZMQ_I_MAILBOX_HPP_INCLUDED__
#include "stdint.hpp"
namespace zmq
{
// Interface to be implemented by mailbox.
class i_mailbox
{
public:
virtual ~i_mailbox () {}
virtual void send (const command_t &cmd_) = 0;
virtual int recv (command_t *cmd_, int timeout_) = 0;
#ifdef HAVE_FORK
// close the file descriptors in the signaller. This is used in a forked
// child process to close the file descriptors so that they do not interfere
// with the context in the parent process.
virtual void forked () = 0;
#endif
};
}
#endif
......@@ -29,11 +29,12 @@
#include "command.hpp"
#include "ypipe.hpp"
#include "mutex.hpp"
#include "i_mailbox.hpp"
namespace zmq
{
class mailbox_t
class mailbox_t : public i_mailbox
{
public:
......
/*
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ 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, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "mailbox_safe.hpp"
#include "clock.hpp"
#include "err.hpp"
zmq::mailbox_safe_t::mailbox_safe_t (mutex_t* sync_) :
sync (sync_)
{
// Get the pipe into passive state. That way, if the users starts by
// polling on the associated file descriptor it will get woken up when
// new command is posted.
const bool ok = cpipe.read (NULL);
zmq_assert (!ok);
}
zmq::mailbox_safe_t::~mailbox_safe_t ()
{
// TODO: Retrieve and deallocate commands inside the cpipe.
// Work around problem that other threads might still be in our
// send() method, by waiting on the mutex before disappearing.
sync->lock ();
sync->unlock ();
}
void zmq::mailbox_safe_t::add_signaler(signaler_t* signaler)
{
signalers.push_back(signaler);
}
void zmq::mailbox_safe_t::remove_signaler(signaler_t* signaler)
{
std::vector<signaler_t*>::iterator it = signalers.begin();
// TODO: make a copy of array and signal outside the lock
for (; it != signalers.end(); ++it){
if (*it == signaler)
break;
}
if (it != signalers.end())
signalers.erase(it);
}
void zmq::mailbox_safe_t::send (const command_t &cmd_)
{
sync->lock ();
cpipe.write (cmd_, false);
const bool ok = cpipe.flush ();
if (!ok) {
cond_var.broadcast ();
for (std::vector<signaler_t*>::iterator it = signalers.begin(); it != signalers.end(); ++it){
(*it)->send();
}
}
sync->unlock ();
}
int zmq::mailbox_safe_t::recv (command_t *cmd_, int timeout_)
{
// Try to get the command straight away.
if (cpipe.read (cmd_))
return 0;
// Wait for signal from the command sender.
int rc = cond_var.wait (sync, timeout_);
if (rc == -1) {
errno_assert (errno == EAGAIN || errno == EINTR);
return -1;
}
// Another thread may already fetch the command
const bool ok = cpipe.read (cmd_);
if (!ok) {
errno = EAGAIN;
return -1;
}
return 0;
}
/*
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ 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, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
void send (const command_t &cmd_);
int recv (command_t *cmd_, int timeout_);
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_MAILBOX_SAFE_HPP_INCLUDED__
#define __ZMQ_MAILBOX_SAFE_HPP_INCLUDED__
#include <vector>
#include <stddef.h>
#include "platform.hpp"
#include "signaler.hpp"
#include "fd.hpp"
#include "config.hpp"
#include "command.hpp"
#include "ypipe.hpp"
#include "mutex.hpp"
#include "i_mailbox.hpp"
#include "condition_variable.hpp"
namespace zmq
{
class mailbox_safe_t : public i_mailbox
{
public:
mailbox_safe_t (mutex_t* sync_);
~mailbox_safe_t ();
void send (const command_t &cmd_);
int recv (command_t *cmd_, int timeout_);
// Add signaler to mailbox which will be called when a message is ready
void add_signaler(signaler_t* signaler);
void remove_signaler(signaler_t* signaler);
#ifdef HAVE_FORK
// close the file descriptors in the signaller. This is used in a forked
// child process to close the file descriptors so that they do not interfere
// with the context in the parent process.
void forked ()
{
// TODO: call fork on the condition variable
}
#endif
private:
// The pipe to store actual commands.
typedef ypipe_t <command_t, command_pipe_granularity> cpipe_t;
cpipe_t cpipe;
// Condition variable to pass signals from writer thread to reader thread.
condition_variable_t cond_var;
// Synchronize access to the mailbox from receivers and senders
mutex_t* sync;
std::vector <zmq::signaler_t* > signalers;
// Disable copying of mailbox_t object.
mailbox_safe_t (const mailbox_safe_t&);
const mailbox_safe_t &operator = (const mailbox_safe_t&);
};
}
#endif
......@@ -60,6 +60,11 @@ namespace zmq
LeaveCriticalSection (&cs);
}
inline CRITICAL_SECTION* get_cs()
{
return &cs;
}
private:
CRITICAL_SECTION cs;
......@@ -83,7 +88,13 @@ namespace zmq
public:
inline mutex_t ()
{
int rc = pthread_mutex_init (&mutex, NULL);
int rc = pthread_mutexattr_init(&attr);
posix_assert (rc);
rc = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
posix_assert (rc);
rc = pthread_mutex_init (&mutex, &attr);
posix_assert (rc);
}
......@@ -91,6 +102,9 @@ namespace zmq
{
int rc = pthread_mutex_destroy (&mutex);
posix_assert (rc);
rc = pthread_mutexattr_destroy (&attr);
posix_assert (rc);
}
inline void lock ()
......@@ -115,9 +129,15 @@ namespace zmq
posix_assert (rc);
}
inline pthread_mutex_t* get_mutex()
{
return &mutex;
}
private:
pthread_mutex_t mutex;
pthread_mutexattr_t attr;
// Disable copy construction and assignment.
mutex_t (const mutex_t&);
......
......@@ -267,6 +267,8 @@ void zmq::pipe_t::process_hiccup (void *pipe_)
outpipe->flush ();
msg_t msg;
while (outpipe->read (&msg)) {
if (!(msg.flags () & msg_t::more))
msgs_written--;
int rc = msg.close ();
errno_assert (rc == 0);
}
......
......@@ -25,7 +25,7 @@
#include "err.hpp"
zmq::server_t::server_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
socket_base_t (parent_, tid_, sid_),
socket_base_t (parent_, tid_, sid_, true),
next_rid (generate_random ())
{
options.type = ZMQ_SERVER;
......
......@@ -54,6 +54,8 @@
#include "ipc_address.hpp"
#include "tcp_address.hpp"
#include "tipc_address.hpp"
#include "mailbox.hpp"
#include "mailbox_safe.hpp"
#ifdef ZMQ_HAVE_OPENPGM
#include "pgm_socket.hpp"
#endif
......@@ -73,6 +75,14 @@
#include "server.hpp"
#include "client.hpp"
#define ENTER_MUTEX() \
if (thread_safe) \
sync.lock();
#define EXIT_MUTEX() \
if (thread_safe) \
sync.unlock();
bool zmq::socket_base_t::check_tag ()
{
return tag == 0xbaddecaf;
......@@ -131,13 +141,16 @@ zmq::socket_base_t *zmq::socket_base_t::create (int type_, class ctx_t *parent_,
}
alloc_assert (s);
if (s->mailbox.get_fd () == retired_fd)
mailbox_t *mailbox = dynamic_cast<mailbox_t*> (s->mailbox);
if (mailbox != NULL && mailbox->get_fd () == retired_fd)
return NULL;
return s;
}
zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_) :
zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_, bool thread_safe_) :
own_t (parent_, tid_),
tag (0xbaddecaf),
ctx_terminated (false),
......@@ -147,22 +160,34 @@ zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_) :
rcvmore (false),
file_desc(-1),
monitor_socket (NULL),
monitor_events (0)
monitor_events (0),
thread_safe (thread_safe_),
reaper_signaler (NULL)
{
options.socket_id = sid_;
options.ipv6 = (parent_->get (ZMQ_IPV6) != 0);
options.linger = parent_->get (ZMQ_BLOCKY)? -1: 0;
if (thread_safe)
mailbox = new mailbox_safe_t(&sync);
else
mailbox = new mailbox_t();
}
zmq::socket_base_t::~socket_base_t ()
{
delete mailbox;
if (reaper_signaler)
delete reaper_signaler;
stop_monitor ();
zmq_assert (destroyed);
}
zmq::mailbox_t *zmq::socket_base_t::get_mailbox ()
zmq::i_mailbox *zmq::socket_base_t::get_mailbox ()
{
return &mailbox;
return mailbox;
}
void zmq::socket_base_t::stop ()
......@@ -275,57 +300,84 @@ void zmq::socket_base_t::attach_pipe (pipe_t *pipe_, bool subscribe_to_all_)
int zmq::socket_base_t::setsockopt (int option_, const void *optval_,
size_t optvallen_)
{
ENTER_MUTEX();
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// First, check whether specific socket type overloads the option.
int rc = xsetsockopt (option_, optval_, optvallen_);
if (rc == 0 || errno != EINVAL)
if (rc == 0 || errno != EINVAL) {
EXIT_MUTEX();
return rc;
}
// If the socket type doesn't support the option, pass it to
// the generic option parser.
return options.setsockopt (option_, optval_, optvallen_);
rc = options.setsockopt (option_, optval_, optvallen_);
EXIT_MUTEX();
return rc;
}
int zmq::socket_base_t::getsockopt (int option_, void *optval_,
size_t *optvallen_)
{
ENTER_MUTEX();
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
if (option_ == ZMQ_RCVMORE) {
if (*optvallen_ < sizeof (int)) {
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
*((int*) optval_) = rcvmore ? 1 : 0;
*optvallen_ = sizeof (int);
EXIT_MUTEX();
return 0;
}
if (option_ == ZMQ_FD) {
if (*optvallen_ < sizeof (fd_t)) {
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
if (thread_safe) {
// thread safe socket doesn't provide file descriptor
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
*((fd_t*) optval_) = mailbox.get_fd ();
*optvallen_ = sizeof (fd_t);
*((fd_t*)optval_) = ((mailbox_t*)mailbox)->get_fd();
*optvallen_ = sizeof(fd_t);
EXIT_MUTEX();
return 0;
}
if (option_ == ZMQ_EVENTS) {
if (*optvallen_ < sizeof (int)) {
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
int rc = process_commands (0, false);
if (rc != 0 && (errno == EINTR || errno == ETERM))
if (rc != 0 && (errno == EINTR || errno == ETERM)) {
EXIT_MUTEX();
return -1;
}
errno_assert (rc == 0);
*((int*) optval_) = 0;
if (has_out ())
......@@ -333,39 +385,51 @@ int zmq::socket_base_t::getsockopt (int option_, void *optval_,
if (has_in ())
*((int*) optval_) |= ZMQ_POLLIN;
*optvallen_ = sizeof (int);
EXIT_MUTEX();
return 0;
}
if (option_ == ZMQ_LAST_ENDPOINT) {
if (*optvallen_ < last_endpoint.size () + 1) {
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
strcpy (static_cast <char *> (optval_), last_endpoint.c_str ());
*optvallen_ = last_endpoint.size () + 1;
EXIT_MUTEX();
return 0;
}
return options.getsockopt (option_, optval_, optvallen_);
int rc = options.getsockopt (option_, optval_, optvallen_);
EXIT_MUTEX();
return rc;
}
int zmq::socket_base_t::bind (const char *addr_)
{
ENTER_MUTEX();
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// Process pending commands, if any.
int rc = process_commands (0, false);
if (unlikely (rc != 0))
if (unlikely (rc != 0)) {
EXIT_MUTEX();
return -1;
}
// Parse addr_ string.
std::string protocol;
std::string address;
if (parse_uri (addr_, protocol, address) || check_protocol (protocol))
if (parse_uri (addr_, protocol, address) || check_protocol (protocol)) {
EXIT_MUTEX();
return -1;
}
if (protocol == "inproc") {
const endpoint_t endpoint = { this, options };
......@@ -374,12 +438,14 @@ int zmq::socket_base_t::bind (const char *addr_)
connect_pending (addr_, this);
last_endpoint.assign (addr_);
}
EXIT_MUTEX();
return rc;
}
if (protocol == "pgm" || protocol == "epgm" || protocol == "norm") {
// For convenience's sake, bind can be used interchageable with
// connect for PGM, EPGM and NORM transports.
EXIT_MUTEX();
return connect (addr_);
}
......@@ -388,6 +454,7 @@ int zmq::socket_base_t::bind (const char *addr_)
io_thread_t *io_thread = choose_io_thread (options.affinity);
if (!io_thread) {
errno = EMTHREAD;
EXIT_MUTEX();
return -1;
}
......@@ -399,6 +466,7 @@ int zmq::socket_base_t::bind (const char *addr_)
if (rc != 0) {
delete listener;
event_bind_failed (address, zmq_errno());
EXIT_MUTEX();
return -1;
}
......@@ -406,6 +474,7 @@ int zmq::socket_base_t::bind (const char *addr_)
listener->get_address (last_endpoint);
add_endpoint (last_endpoint.c_str (), (own_t *) listener, NULL);
EXIT_MUTEX();
return 0;
}
......@@ -418,6 +487,7 @@ int zmq::socket_base_t::bind (const char *addr_)
if (rc != 0) {
delete listener;
event_bind_failed (address, zmq_errno());
EXIT_MUTEX();
return -1;
}
......@@ -425,6 +495,7 @@ int zmq::socket_base_t::bind (const char *addr_)
listener->get_address (last_endpoint);
add_endpoint (last_endpoint.c_str (), (own_t *) listener, NULL);
EXIT_MUTEX();
return 0;
}
#endif
......@@ -437,6 +508,7 @@ int zmq::socket_base_t::bind (const char *addr_)
if (rc != 0) {
delete listener;
event_bind_failed (address, zmq_errno());
EXIT_MUTEX();
return -1;
}
......@@ -444,31 +516,40 @@ int zmq::socket_base_t::bind (const char *addr_)
listener->get_address (last_endpoint);
add_endpoint (addr_, (own_t *) listener, NULL);
EXIT_MUTEX();
return 0;
}
#endif
EXIT_MUTEX();
zmq_assert (false);
return -1;
}
int zmq::socket_base_t::connect (const char *addr_)
{
ENTER_MUTEX();
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// Process pending commands, if any.
int rc = process_commands (0, false);
if (unlikely (rc != 0))
if (unlikely (rc != 0)) {
EXIT_MUTEX();
return -1;
}
// Parse addr_ string.
std::string protocol;
std::string address;
if (parse_uri (addr_, protocol, address) || check_protocol (protocol))
if (parse_uri (addr_, protocol, address) || check_protocol (protocol)) {
EXIT_MUTEX();
return -1;
}
if (protocol == "inproc") {
......@@ -566,6 +647,7 @@ int zmq::socket_base_t::connect (const char *addr_)
// remember inproc connections for disconnect
inprocs.insert (inprocs_t::value_type (std::string (addr_), new_pipes [0]));
EXIT_MUTEX();
return 0;
}
bool is_single_connect = (options.type == ZMQ_DEALER ||
......@@ -577,6 +659,7 @@ int zmq::socket_base_t::connect (const char *addr_)
// There is no valid use for multiple connects for SUB-PUB nor
// DEALER-ROUTER nor REQ-REP. Multiple connects produces
// nonsensical results.
EXIT_MUTEX();
return 0;
}
}
......@@ -585,6 +668,7 @@ int zmq::socket_base_t::connect (const char *addr_)
io_thread_t *io_thread = choose_io_thread (options.affinity);
if (!io_thread) {
errno = EMTHREAD;
EXIT_MUTEX();
return -1;
}
......@@ -624,6 +708,7 @@ int zmq::socket_base_t::connect (const char *addr_)
if (rc == -1) {
errno = EINVAL;
delete paddr;
EXIT_MUTEX();
return -1;
}
// Defer resolution until a socket is opened
......@@ -637,6 +722,7 @@ int zmq::socket_base_t::connect (const char *addr_)
int rc = paddr->resolved.ipc_addr->resolve (address.c_str ());
if (rc != 0) {
delete paddr;
EXIT_MUTEX();
return -1;
}
}
......@@ -652,6 +738,7 @@ int zmq::socket_base_t::connect (const char *addr_)
if (res != NULL)
pgm_freeaddrinfo (res);
if (rc != 0 || port_number == 0)
EXIT_MUTEX();
return -1;
}
#endif
......@@ -663,6 +750,7 @@ int zmq::socket_base_t::connect (const char *addr_)
int rc = paddr->resolved.tipc_addr->resolve (address.c_str());
if (rc != 0) {
delete paddr;
EXIT_MUTEX();
return -1;
}
}
......@@ -708,6 +796,7 @@ int zmq::socket_base_t::connect (const char *addr_)
paddr->to_string (last_endpoint);
add_endpoint (addr_, (own_t *) session, newpipe);
EXIT_MUTEX();
return 0;
}
......@@ -720,43 +809,55 @@ void zmq::socket_base_t::add_endpoint (const char *addr_, own_t *endpoint_, pipe
int zmq::socket_base_t::term_endpoint (const char *addr_)
{
ENTER_MUTEX();
// Check whether the library haven't been shut down yet.
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// Check whether endpoint address passed to the function is valid.
if (unlikely (!addr_)) {
errno = EINVAL;
EXIT_MUTEX();
return -1;
}
// Process pending commands, if any, since there could be pending unprocessed process_own()'s
// (from launch_child() for example) we're asked to terminate now.
int rc = process_commands (0, false);
if (unlikely (rc != 0))
if (unlikely(rc != 0)) {
EXIT_MUTEX();
return -1;
}
// Parse addr_ string.
std::string protocol;
std::string address;
if (parse_uri (addr_, protocol, address) || check_protocol (protocol))
if (parse_uri(addr_, protocol, address) || check_protocol(protocol)) {
EXIT_MUTEX();
return -1;
}
// Disconnect an inproc socket
if (protocol == "inproc") {
if (unregister_endpoint (std::string (addr_), this) == 0)
if (unregister_endpoint (std::string(addr_), this) == 0) {
EXIT_MUTEX();
return 0;
}
std::pair <inprocs_t::iterator, inprocs_t::iterator> range = inprocs.equal_range (std::string (addr_));
if (range.first == range.second) {
errno = ENOENT;
EXIT_MUTEX();
return -1;
}
for (inprocs_t::iterator it = range.first; it != range.second; ++it)
it->second->terminate (true);
inprocs.erase (range.first, range.second);
EXIT_MUTEX();
return 0;
}
......@@ -764,6 +865,7 @@ int zmq::socket_base_t::term_endpoint (const char *addr_)
std::pair <endpoints_t::iterator, endpoints_t::iterator> range = endpoints.equal_range (std::string (addr_));
if (range.first == range.second) {
errno = ENOENT;
EXIT_MUTEX();
return -1;
}
......@@ -774,27 +876,34 @@ int zmq::socket_base_t::term_endpoint (const char *addr_)
term_child (it->second.first);
}
endpoints.erase (range.first, range.second);
EXIT_MUTEX();
return 0;
}
int zmq::socket_base_t::send (msg_t *msg_, int flags_)
{
ENTER_MUTEX();
// Check whether the library haven't been shut down yet.
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// Check whether message passed to the function is valid.
if (unlikely (!msg_ || !msg_->check ())) {
errno = EFAULT;
EXIT_MUTEX();
return -1;
}
// Process pending commands, if any.
int rc = process_commands (0, true);
if (unlikely (rc != 0))
if (unlikely (rc != 0)) {
EXIT_MUTEX();
return -1;
}
// Clear any user-visible flags that are set on the message.
msg_->reset_flags (msg_t::more);
......@@ -807,15 +916,21 @@ int zmq::socket_base_t::send (msg_t *msg_, int flags_)
// Try to send the message.
rc = xsend (msg_);
if (rc == 0)
if (rc == 0) {
EXIT_MUTEX();
return 0;
if (unlikely (errno != EAGAIN))
}
if (unlikely (errno != EAGAIN)) {
EXIT_MUTEX();
return -1;
}
// In case of non-blocking send we'll simply propagate
// the error - including EAGAIN - up the stack.
if (flags_ & ZMQ_DONTWAIT || options.sndtimeo == 0)
if (flags_ & ZMQ_DONTWAIT || options.sndtimeo == 0) {
EXIT_MUTEX();
return -1;
}
// Compute the time when the timeout should occur.
// If the timeout is infinite, don't care.
......@@ -826,35 +941,46 @@ int zmq::socket_base_t::send (msg_t *msg_, int flags_)
// command, process it and try to send the message again.
// If timeout is reached in the meantime, return EAGAIN.
while (true) {
if (unlikely (process_commands (timeout, false) != 0))
if (unlikely (process_commands (timeout, false) != 0)) {
EXIT_MUTEX();
return -1;
}
rc = xsend (msg_);
if (rc == 0)
break;
if (unlikely (errno != EAGAIN))
if (unlikely (errno != EAGAIN)) {
EXIT_MUTEX();
return -1;
}
if (timeout > 0) {
timeout = (int) (end - clock.now_ms ());
if (timeout <= 0) {
errno = EAGAIN;
EXIT_MUTEX();
return -1;
}
}
}
EXIT_MUTEX();
return 0;
}
int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
{
ENTER_MUTEX();
// Check whether the library haven't been shut down yet.
if (unlikely (ctx_terminated)) {
errno = ETERM;
EXIT_MUTEX();
return -1;
}
// Check whether message passed to the function is valid.
if (unlikely (!msg_ || !msg_->check ())) {
errno = EFAULT;
EXIT_MUTEX();
return -1;
}
......@@ -867,21 +993,26 @@ int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
// described above) from the one used by 'send'. This is because counting
// ticks is more efficient than doing RDTSC all the time.
if (++ticks == inbound_poll_rate) {
if (unlikely (process_commands (0, false) != 0))
if (unlikely (process_commands (0, false) != 0)) {
EXIT_MUTEX();
return -1;
}
ticks = 0;
}
// Get the message.
int rc = xrecv (msg_);
if (unlikely (rc != 0 && errno != EAGAIN))
if (unlikely (rc != 0 && errno != EAGAIN)) {
EXIT_MUTEX();
return -1;
}
// If we have the message, return immediately.
if (rc == 0) {
if (file_desc != retired_fd)
msg_->set_fd(file_desc);
extract_flags (msg_);
EXIT_MUTEX();
return 0;
}
......@@ -890,16 +1021,22 @@ int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
// activate_reader command already waiting int a command pipe.
// If it's not, return EAGAIN.
if (flags_ & ZMQ_DONTWAIT || options.rcvtimeo == 0) {
if (unlikely (process_commands (0, false) != 0))
if (unlikely (process_commands (0, false) != 0)) {
EXIT_MUTEX();
return -1;
}
ticks = 0;
rc = xrecv (msg_);
if (rc < 0)
if (rc < 0) {
EXIT_MUTEX();
return rc;
}
if (file_desc != retired_fd)
msg_->set_fd(file_desc);
extract_flags (msg_);
EXIT_MUTEX();
return 0;
}
......@@ -912,20 +1049,25 @@ int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
// we are able to fetch a message.
bool block = (ticks != 0);
while (true) {
if (unlikely (process_commands (block ? timeout : 0, false) != 0))
if (unlikely (process_commands (block ? timeout : 0, false) != 0)) {
EXIT_MUTEX();
return -1;
}
rc = xrecv (msg_);
if (rc == 0) {
ticks = 0;
break;
}
if (unlikely (errno != EAGAIN))
if (unlikely (errno != EAGAIN)) {
EXIT_MUTEX();
return -1;
}
block = true;
if (timeout > 0) {
timeout = (int) (end - clock.now_ms ());
if (timeout <= 0) {
errno = EAGAIN;
EXIT_MUTEX();
return -1;
}
}
......@@ -934,6 +1076,7 @@ int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
if (file_desc != retired_fd)
msg_->set_fd(file_desc);
extract_flags (msg_);
EXIT_MUTEX();
return 0;
}
......@@ -964,7 +1107,27 @@ void zmq::socket_base_t::start_reaping (poller_t *poller_)
{
// Plug the socket to the reaper thread.
poller = poller_;
handle = poller->add_fd (mailbox.get_fd (), this);
fd_t fd;
if (!thread_safe)
fd = ((mailbox_t*)mailbox)->get_fd();
else {
ENTER_MUTEX();
reaper_signaler = new signaler_t();
// Add signaler to the safe mailbox
fd = reaper_signaler->get_fd();
((mailbox_safe_t*)mailbox)->add_signaler(reaper_signaler);
// Send a signal to make sure reaper handle existing commands
reaper_signaler->send();
EXIT_MUTEX();
}
handle = poller->add_fd (fd, this);
poller->set_pollin (handle);
// Initialise the termination and check whether it can be deallocated
......@@ -980,7 +1143,7 @@ int zmq::socket_base_t::process_commands (int timeout_, bool throttle_)
if (timeout_ != 0) {
// If we are asked to wait, simply ask mailbox to wait.
rc = mailbox.recv (&cmd, timeout_);
rc = mailbox->recv (&cmd, timeout_);
}
else {
......@@ -1007,13 +1170,13 @@ int zmq::socket_base_t::process_commands (int timeout_, bool throttle_)
}
// Check whether there are any commands pending for this thread.
rc = mailbox.recv (&cmd, 0);
rc = mailbox->recv (&cmd, 0);
}
// Process all available commands.
while (rc == 0) {
cmd.destination->process_command (cmd);
rc = mailbox.recv (&cmd, 0);
rc = mailbox->recv (&cmd, 0);
}
if (errno == EINTR)
......@@ -1118,7 +1281,15 @@ void zmq::socket_base_t::in_event ()
// of the reaper thread. Process any commands from other threads/sockets
// that may be available at the moment. Ultimately, the socket will
// be destroyed.
ENTER_MUTEX();
// If the socket is thread safe we need to unsignal the reaper signaler
if (thread_safe)
reaper_signaler->recv();
process_commands (0, false);
EXIT_MUTEX();
check_destroy ();
}
......
......@@ -31,7 +31,7 @@
#include "poller.hpp"
#include "atomic_counter.hpp"
#include "i_poll_events.hpp"
#include "mailbox.hpp"
#include "i_mailbox.hpp"
#include "stdint.hpp"
#include "clock.hpp"
#include "pipe.hpp"
......@@ -66,7 +66,7 @@ namespace zmq
uint32_t tid_, int sid_);
// Returns the mailbox associated with this socket.
mailbox_t *get_mailbox ();
i_mailbox *get_mailbox ();
// Interrupt blocking call if the socket is stuck in one.
// This function can be called from a different thread!
......@@ -123,7 +123,7 @@ namespace zmq
protected:
socket_base_t (zmq::ctx_t *parent_, uint32_t tid_, int sid_);
socket_base_t (zmq::ctx_t *parent_, uint32_t tid_, int sid_, bool thread_safe_ = false);
virtual ~socket_base_t ();
// Concrete algorithms for the x- methods are to be defined by
......@@ -223,7 +223,7 @@ namespace zmq
void process_term (int linger_);
// Socket's mailbox object.
mailbox_t mailbox;
i_mailbox* mailbox;
// List of attached pipes.
typedef array_t <pipe_t, 3> pipes_t;
......@@ -257,9 +257,17 @@ namespace zmq
// Last socket endpoint resolved URI
std::string last_endpoint;
socket_base_t (const socket_base_t&);
const socket_base_t &operator = (const socket_base_t&);
// Indicate if the socket is thread safe
bool thread_safe;
// Signaler to be used in the reaping stage
signaler_t* reaper_signaler;
// Mutex for synchronize access to the socket in thread safe mode
mutex_t sync;
socket_base_t (const socket_base_t&);
const socket_base_t &operator = (const socket_base_t&);
};
}
......
......@@ -176,10 +176,10 @@ int zmq::stream_t::xsetsockopt (int option_, const void *optval_,
return 0;
}
break;
case ZMQ_STREAM_NOTIFY:
if (is_int && (value == 0 || value == 1)) {
options.raw_notify = value;
options.raw_notify = (value != 0);
return 0;
}
break;
......@@ -221,6 +221,12 @@ int zmq::stream_t::xrecv (msg_t *msg_)
blob_t identity = pipe->get_identity ();
rc = msg_->init_size (identity.size ());
errno_assert (rc == 0);
// forward metadata (if any)
metadata_t *metadata = prefetched_msg.metadata();
if (metadata)
msg_->set_metadata(metadata);
memcpy (msg_->data (), identity.data (), identity.size ());
msg_->set_flags (msg_t::more);
......@@ -249,6 +255,12 @@ bool zmq::stream_t::xhas_in ()
blob_t identity = pipe->get_identity ();
rc = prefetched_id.init_size (identity.size ());
errno_assert (rc == 0);
// forward metadata (if any)
metadata_t *metadata = prefetched_msg.metadata();
if (metadata)
prefetched_id.set_metadata(metadata);
memcpy (prefetched_id.data (), identity.data (), identity.size ());
prefetched_id.set_flags (msg_t::more);
......@@ -277,7 +289,7 @@ void zmq::stream_t::identify_peer (pipe_t *pipe_)
connect_rid.length ());
connect_rid.clear ();
outpipes_t::iterator it = outpipes.find (identity);
if (it != outpipes.end ())
if (it != outpipes.end ())
zmq_assert(false);
}
else {
......
......@@ -36,7 +36,6 @@
#include <string.h>
#include <new>
#include <sstream>
#include <iostream>
#include "stream_engine.hpp"
#include "io_thread.hpp"
......@@ -192,14 +191,21 @@ void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
handshaking = false;
next_msg = &stream_engine_t::pull_msg_from_session;
process_msg = &stream_engine_t::push_msg_to_session;
process_msg = &stream_engine_t::push_raw_msg_to_session;
properties_t properties;
if (init_properties(properties)) {
// Compile metadata.
zmq_assert (metadata == NULL);
metadata = new (std::nothrow) metadata_t (properties);
}
if (options.raw_notify) {
// For raw sockets, send an initial 0-length message to the
// application so that it knows a peer has connected.
msg_t connector;
connector.init();
push_msg_to_session (&connector);
push_raw_msg_to_session (&connector);
connector.close();
session->flush ();
}
......@@ -804,13 +810,8 @@ void zmq::stream_engine_t::mechanism_ready ()
process_msg = &stream_engine_t::write_credential;
// Compile metadata.
typedef metadata_t::dict_t properties_t;
properties_t properties;
// If we have a peer_address, add it to metadata
if (!peer_address.empty()) {
properties.insert(std::make_pair("Peer-Address", peer_address));
}
init_properties(properties);
// Add ZAP properties.
const properties_t& zap_properties = mechanism->get_zap_properties ();
......@@ -835,6 +836,12 @@ int zmq::stream_engine_t::push_msg_to_session (msg_t *msg_)
return session->push_msg (msg_);
}
int zmq::stream_engine_t::push_raw_msg_to_session (msg_t *msg_) {
if (metadata)
msg_->set_metadata(metadata);
return push_msg_to_session(msg_);
}
int zmq::stream_engine_t::write_credential (msg_t *msg_)
{
zmq_assert (mechanism != NULL);
......@@ -938,6 +945,12 @@ void zmq::stream_engine_t::set_handshake_timer ()
}
}
bool zmq::stream_engine_t::init_properties (properties_t & properties) {
if (peer_address.empty()) return false;
properties.insert (std::make_pair("Peer-Address", peer_address));
return true;
}
void zmq::stream_engine_t::timer_event (int id_)
{
zmq_assert (id_ == handshake_timer_id);
......
......@@ -59,7 +59,7 @@ namespace zmq
timeout_error
};
stream_engine_t (fd_t fd_, const options_t &options_,
stream_engine_t (fd_t fd_, const options_t &options_,
const std::string &endpoint);
~stream_engine_t ();
......@@ -77,7 +77,6 @@ namespace zmq
void timer_event (int id_);
private:
// Unplug the engine from the session.
void unplug ();
......@@ -99,6 +98,8 @@ namespace zmq
int pull_msg_from_session (msg_t *msg_);
int push_msg_to_session (msg_t *msg);
int push_raw_msg_to_session (msg_t *msg);
int write_credential (msg_t *msg_);
int pull_and_encode (msg_t *msg_);
int decode_and_push (msg_t *msg_);
......@@ -113,6 +114,9 @@ namespace zmq
void set_handshake_timer();
typedef metadata_t::dict_t properties_t;
bool init_properties (properties_t & properties);
// Underlying socket.
fd_t s;
......
......@@ -627,8 +627,7 @@ int zmq_msg_get (zmq_msg_t *msg_, int property_)
case ZMQ_MORE:
return (((zmq::msg_t*) msg_)->flags () & zmq::msg_t::more)? 1: 0;
case ZMQ_SRCFD:
// warning: int64_t to int
return ((zmq::msg_t*) msg_)->fd ();
return (int)((zmq::msg_t*) msg_)->fd ();
case ZMQ_SHARED:
return (((zmq::msg_t*) msg_)->is_cmsg ()) ||
(((zmq::msg_t*) msg_)->flags () & zmq::msg_t::shared)? 1: 0;
......
......@@ -45,6 +45,8 @@ set(tests
test_connect_rid
test_xpub_nodrop
test_pub_invert_matching
test_thread_safe
test_client_server
)
if(NOT WIN32)
list(APPEND tests
......
......@@ -80,17 +80,32 @@ test_stream_to_dealer (void)
assert (rc > 0);
assert (zmq_msg_more (&identity));
// Verify the existence of Peer-Address metadata
char const* peer_address = zmq_msg_gets (&identity, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
// Second frame is zero
byte buffer [255];
rc = zmq_recv (stream, buffer, 255, 0);
assert (rc == 0);
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&identity, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
// Real data follows
// First frame is identity
rc = zmq_msg_recv (&identity, stream, 0);
assert (rc > 0);
assert (zmq_msg_more (&identity));
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&identity, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
// Second frame is greeting signature
rc = zmq_recv (stream, buffer, 255, 0);
assert (rc == 10);
......@@ -182,7 +197,7 @@ test_stream_to_stream (void)
// Set-up our context and sockets
void *ctx = zmq_ctx_new ();
assert (ctx);
void *server = zmq_socket (ctx, ZMQ_STREAM);
assert (server);
int enabled = 1;
......@@ -200,7 +215,7 @@ test_stream_to_stream (void)
uint8_t id [256];
size_t id_size = 256;
uint8_t buffer [256];
// Connecting sends a zero message
// Server: First frame is identity, second frame is zero
id_size = zmq_recv (server, id, 256, 0);
......@@ -223,19 +238,19 @@ test_stream_to_stream (void)
// Second frame is HTTP GET request
rc = zmq_send (client, "GET /\n\n", 7, 0);
assert (rc == 7);
// Get HTTP request; ID frame and then request
id_size = zmq_recv (server, id, 256, 0);
assert (id_size > 0);
rc = zmq_recv (server, buffer, 256, 0);
assert (rc != -1);
assert (memcmp (buffer, "GET /\n\n", 7) == 0);
// Send reply back to client
char http_response [] =
"HTTP/1.0 200 OK\r\n"
"Content-Type: text/plain\r\n"
"\r\n"
"HTTP/1.0 200 OK\r\n"
"Content-Type: text/plain\r\n"
"\r\n"
"Hello, World!";
rc = zmq_send (server, id, id_size, ZMQ_SNDMORE);
assert (rc != -1);
......
/*:
Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ 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, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "testutil.hpp"
void worker1(void* s);
void worker2(void* s);
int main (void)
{
setup_test_environment();
void *ctx = zmq_ctx_new ();
assert (ctx);
void *client = zmq_socket (ctx, ZMQ_CLIENT);
void *client2 = zmq_socket (ctx, ZMQ_CLIENT);
int rc;
rc = zmq_bind (client, "tcp://127.0.0.1:5560");
assert (rc == 0);
rc = zmq_connect (client2, "tcp://127.0.0.1:5560");
assert (rc == 0);
void* t1 = zmq_threadstart(worker1, client2);
void* t2 = zmq_threadstart(worker2, client2);
char data[1];
data[0] = 0;
for (int i=0; i < 10; i++) {
rc = zmq_send_const(client, data, 1, 0);
assert (rc == 1);
rc = zmq_send_const(client, data, 1, 0);
assert(rc == 1);
char a, b;
rc = zmq_recv(client, &a, 1, 0);
assert(rc == 1);
rc = zmq_recv(client, &b, 1, 0);
assert(rc == 1);
// make sure they came from different threads
assert((a == 1 && b == 2) || (a == 2 && b == 1));
}
// make the thread exit
data[0] = 1;
rc = zmq_send_const(client, data, 1, 0);
assert (rc == 1);
rc = zmq_send_const(client, data, 1, 0);
assert(rc == 1);
zmq_threadclose(t1);
zmq_threadclose(t2);
rc = zmq_close (client2);
assert (rc == 0);
rc = zmq_close (client);
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
return 0 ;
}
void worker1(void* s)
{
const char worker_id = 1;
char c;
while (true)
{
int rc = zmq_recv(s, &c,1, 0);
assert(rc == 1);
if (c == 0)
{
msleep(10);
rc = zmq_send_const(s,&worker_id, 1, 0);
assert(rc == 1);
}
else
{
// we got exit request
break;
}
}
}
void worker2(void* s)
{
const char worker_id = 2;
char c;
while (true)
{
int rc = zmq_recv(s, &c,1, 0);
assert(rc == 1);
assert(c == 1 || c == 0);
if (c == 0)
{
msleep(10);
rc = zmq_send_const(s,&worker_id, 1, 0);
assert(rc == 1);
}
else
{
// we got exit request
break;
}
}
}
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