diff --git a/configure.ac b/configure.ac index 0c371aaa..b3943298 100644 --- a/configure.ac +++ b/configure.ac @@ -1,736 +1,737 @@ # -*- Autoconf -*- # Process this file with autoconf to produce a configure script. # bootstrap / init AC_PREREQ([2.61]) AC_INIT([corosync], m4_esyscmd([build-aux/git-version-gen .tarball-version]), [discuss@lists.corosync.org]) AC_USE_SYSTEM_EXTENSIONS AM_INIT_AUTOMAKE([-Wno-portability 1.11]) LT_PREREQ([2.2.6]) LT_INIT AM_SILENT_RULES([yes]) AC_CONFIG_SRCDIR([lib/cpg.c]) AC_CONFIG_HEADER([include/corosync/config.h]) AC_CONFIG_MACRO_DIR([m4]) AC_CANONICAL_HOST AC_LANG([C]) AC_SUBST(WITH_LIST, [""]) #Enable inter-library dependencies AC_ARG_ENABLE(interlib-deps, [AC_HELP_STRING([--disable-interlib-deps ],[disable inter-library dependencies (might break builds)])], [enable_interlib_deps="$enableval"], [enable_interlib_deps="yes"]) AC_MSG_NOTICE([enable inter-library dependencies: $enable_interlib_deps]) if test "${enable_interlib_deps}" == "yes"; then link_all_deplibs=yes link_all_deplibs_CXX=yes else link_all_deplibs=no link_all_deplibs_CXX=no fi dnl Fix default variables - "prefix" variable if not specified if test "$prefix" = "NONE"; then prefix="/usr" dnl Fix "localstatedir" variable if not specified if test "$localstatedir" = "\${prefix}/var"; then localstatedir="/var" fi dnl Fix "sysconfdir" variable if not specified if test "$sysconfdir" = "\${prefix}/etc"; then sysconfdir="/etc" fi dnl Fix "libdir" variable if not specified if test "$libdir" = "\${exec_prefix}/lib"; then if test -e /usr/lib64; then libdir="/usr/lib64" else libdir="/usr/lib" fi fi fi if test "$srcdir" = "."; then AC_MSG_NOTICE([building in place srcdir:$srcdir]) AC_DEFINE([BUILDING_IN_PLACE], 1, [building in place]) else AC_MSG_NOTICE([building out of tree srcdir:$srcdir]) fi # Checks for programs. # check stolen from gnulib/m4/gnu-make.m4 if ! ${MAKE-make} --version /cannot/make/this >/dev/null 2>&1; then AC_MSG_ERROR([you don't seem to have GNU make; it is required]) fi sinclude(coroysync-default.m4) AC_PROG_CC AC_PROG_CXX AC_PROG_INSTALL AC_PROG_LN_S AC_PROG_MAKE_SET AC_PROG_RANLIB AC_PROG_SED AC_PROG_LIBTOOL AC_CHECK_PROGS([GROFF], [groff]) AC_CHECK_PROGS([PKGCONFIG], [pkg-config]) AC_CHECK_PROGS([AUGTOOL], [augtool]) AC_CHECK_PROGS([DOT], [dot]) AC_CHECK_PROGS([DOXYGEN], [doxygen]) AC_CHECK_PROGS([AWK], [awk]) AC_CHECK_PROGS([SED], [sed]) AC_PATH_PROG([BASHPATH], [bash]) # Checks for compiler characteristics. AC_PROG_GCC_TRADITIONAL AC_C_CONST AC_C_INLINE AC_C_VOLATILE # Checks for header files. AC_HEADER_DIRENT AC_HEADER_STDC AC_HEADER_SYS_WAIT AC_CHECK_HEADERS([arpa/inet.h fcntl.h limits.h netdb.h netinet/in.h stdint.h \ stdlib.h string.h sys/ioctl.h sys/param.h sys/socket.h \ sys/time.h syslog.h unistd.h sys/types.h getopt.h malloc.h \ utmpx.h ifaddrs.h stddef.h sys/file.h sys/uio.h]) # Check entries in specific structs AC_CHECK_MEMBER([struct sockaddr_in.sin_len], [AC_DEFINE_UNQUOTED([HAVE_SOCK_SIN_LEN], [1], [sockaddr_in needs sin_len])], [], [[#include ]]) AC_CHECK_MEMBER([struct sockaddr_in6.sin6_len], [AC_DEFINE_UNQUOTED([HAVE_SOCK_SIN6_LEN], [1], [sockaddr_in6 needs sin6_len])], [], [[#include ]]) AC_CHECK_MEMBER([struct msghdr.msg_control], [AC_DEFINE_UNQUOTED([HAVE_MSGHDR_CONTROL], [1], [msghdr has msg_control])], [], [[#include ]]) AC_CHECK_MEMBER([struct msghdr.msg_controllen], [AC_DEFINE_UNQUOTED([HAVE_MSGHDR_CONTROLLEN], [1], [msghdr has msg_controllen])], [], [[#include ]]) AC_CHECK_MEMBER([struct msghdr.msg_flags], [AC_DEFINE_UNQUOTED([HAVE_MSGHDR_FLAGS], [1], [msghdr has msg_flags])], [], [[#include ]]) AC_CHECK_MEMBER([struct msghdr.msg_accrights], [AC_DEFINE_UNQUOTED([HAVE_MSGHDR_ACCRIGHTS], [1], [msghdr has msg_accrights])], [], [[#include ]]) AC_CHECK_MEMBER([struct msghdr.msg_accrightslen], [AC_DEFINE_UNQUOTED([HAVE_MSGHDR_ACCRIGHTSLEN], [1], [msghdr has msg_accrightslen])], [], [[#include ]]) # Checks for typedefs. AC_TYPE_UID_T AC_TYPE_INT16_T AC_TYPE_INT32_T AC_TYPE_INT64_T AC_TYPE_INT8_T AC_TYPE_UINT16_T AC_TYPE_UINT32_T AC_TYPE_UINT64_T AC_TYPE_UINT8_T AC_TYPE_SIZE_T AC_TYPE_SSIZE_T AC_TYPE_SIGNAL # Checks for libraries. PKG_CHECK_MODULES([nss],[nss]) PKG_CHECK_MODULES([LIBQB], [libqb]) AC_CHECK_LIB([qb], [qb_log_thread_priority_set], \ have_qb_log_thread_priority_set="yes", \ have_qb_log_thread_priority_set="no") if test "x${have_qb_log_thread_priority_set}" = xyes; then AC_DEFINE_UNQUOTED([HAVE_QB_LOG_THREAD_PRIORITY_SET], 1, [have qb_log_thread_priority_set]) fi AC_CHECK_LIB([pthread], [pthread_create]) AC_CHECK_LIB([socket], [socket]) AC_CHECK_LIB([nsl], [t_open]) AC_CHECK_LIB([rt], [sched_getscheduler]) +AC_CHECK_LIB([z], [crc32]) # Checks for library functions. AC_FUNC_ALLOCA AC_FUNC_CLOSEDIR_VOID AC_FUNC_ERROR_AT_LINE AC_FUNC_FORK AC_FUNC_MALLOC AC_FUNC_MEMCMP AC_FUNC_MMAP AC_FUNC_REALLOC AC_FUNC_SELECT_ARGTYPES AC_FUNC_VPRINTF AC_CHECK_FUNCS([alarm alphasort atexit bzero dup2 endgrent endpwent fcntl \ getcwd getpeerucred getpeereid gettimeofday inet_ntoa memmove \ memset mkdir scandir select socket strcasecmp strchr strdup \ strerror strrchr strspn strstr pthread_setschedparam \ sched_get_priority_max sched_setscheduler getifaddrs \ clock_gettime ftruncate gethostname localtime_r munmap strtol]) AC_CONFIG_FILES([Makefile exec/Makefile include/Makefile init/Makefile lib/Makefile common_lib/Makefile man/Makefile pkgconfig/Makefile test/Makefile cts/Makefile cts/agents/Makefile cts/CTSvars.py tools/Makefile conf/Makefile qdevices/Makefile Doxyfile]) ### Local business dnl =============================================== dnl Functions / global M4 variables dnl =============================================== dnl Global list of LIB names m4_define([local_soname_list], [])dnl dnl Upcase parameter m4_define([local_upcase], [translit([$*], [a-z], [A-Z])])dnl dnl M4 macro for include lib/lib$1.soname and subst that m4_define([LIB_SONAME_IMPORT],[dnl m4_define([local_libname], local_upcase($1)[_SONAME])dnl m4_define([local_soname], translit(m4_sinclude(lib/lib$1.verso), [ ], []))dnl local_libname="local_soname"dnl m4_define([local_soname_list], m4_defn([local_soname_list])[,]local_libname[,]local_upcase($1))dnl AC_SUBST(local_libname)dnl ])dnl dnl M4 macro for print padspaces (used in LIB_MSG_RESULT). It takes 2 arguments, length of string to pad and desired dnl (padded) length m4_define([m4_printpadspace],[ifelse(m4_eval([$2 - $1 < 1]),[1],,[ ][m4_printpadspace([$1],m4_eval([$2 - 1]))])])dnl dnl Show AC_MSG_RESULT for specific libraries m4_define([LIB_MSG_RESULT], [ifelse([$#], [1], ,[dnl AC_MSG_RESULT([ $2 Library SONAME m4_printpadspace(len($2),8) = ${$1}]) LIB_MSG_RESULT(m4_shift(m4_shift($@)))dnl ])])dnl # =============================================== # Helpers # =============================================== ## helper for CC stuff cc_supports_flag() { BACKUP="$CPPFLAGS" CPPFLAGS="$CPPFLAGS $@" AC_MSG_CHECKING([whether $CC supports "$@"]) AC_PREPROC_IFELSE([AC_LANG_PROGRAM([])], [RC=0; AC_MSG_RESULT([yes])], [RC=1; AC_MSG_RESULT([no])]) CPPFLAGS="$BACKUP" return $RC } ## cleanup AC_MSG_NOTICE(Sanitizing prefix: ${prefix}) case $prefix in NONE) prefix=/usr/local;; esac AC_MSG_NOTICE(Sanitizing exec_prefix: ${exec_prefix}) case $exec_prefix in dnl For consistency with Corosync, map NONE->$prefix NONE) exec_prefix=$prefix;; prefix) exec_prefix=$prefix;; esac ## local defines PACKAGE_FEATURES="" LINT_FLAGS="-weak -unrecog +posixlib +ignoresigns -fcnuse \ -badflag -D__gnuc_va_list=va_list -D__attribute\(x\)=" # default libraries SONAME SOMAJOR="5" SOMINOR="0" SOMICRO="0" SONAME="${SOMAJOR}.${SOMINOR}.${SOMICRO}" # specific libraries SONAME LIB_SONAME_IMPORT([cfg]) LIB_SONAME_IMPORT([cpg]) LIB_SONAME_IMPORT([quorum]) LIB_SONAME_IMPORT([sam]) LIB_SONAME_IMPORT([votequorum]) LIB_SONAME_IMPORT([cmap]) # local options AC_ARG_ENABLE([ansi], [ --enable-ansi : force to build with ANSI standards. ], [ default="no" ]) AC_ARG_ENABLE([fatal-warnings], [ --enable-fatal-warnings : enable fatal warnings. ], [ default="no" ]) AC_ARG_ENABLE([debug], [ --enable-debug : enable debug build. ], [ default="no" ]) AC_ARG_ENABLE([secure-build], [ --enable-secure-build : enable PIE/RELRO build. ], [], [enable_secure_build="yes"]) AC_ARG_ENABLE([user-flags], [ --enable-user-flags : rely on user environment. ], [ default="no" ]) AC_ARG_ENABLE([coverage], [ --enable-coverage : coverage analysis of the codebase. ], [ default="no" ]) AC_ARG_ENABLE([small-memory-footprint], [ --enable-small-memory-footprint : Use small message queues and small messages sizes. ], [ default="no" ]) AC_ARG_ENABLE([dbus], [ --enable-dbus : dbus events. ],, [ enable_dbus="no" ]) AC_ARG_ENABLE([testagents], [ --enable-testagents : Install Test Agents. ],, [ default="no" ]) AC_ARG_ENABLE([rdma], [ --enable-rdma : Infiniband RDMA transport support ],, [ enable_rdma="no" ]) AM_CONDITIONAL(BUILD_RDMA, test x$enable_rdma = xyes) AC_ARG_ENABLE([monitoring], [ --enable-monitoring : resource monitoring ],, [ default="no" ]) AM_CONDITIONAL(BUILD_MONITORING, test x$enable_monitoring = xyes) AC_ARG_ENABLE([watchdog], [ --enable-watchdog : Watchdog support ],, [ default="no" ]) AM_CONDITIONAL(BUILD_WATCHDOG, test x$enable_watchdog = xyes) AC_ARG_ENABLE([augeas], [ --enable-augeas : Install the augeas lens for corosync.conf ],, [ enable_augeas="no" ]) AM_CONDITIONAL(INSTALL_AUGEAS, test x$enable_augeas = xyes) AC_ARG_ENABLE([systemd], [ --enable-systemd : Install systemd service files],, [ enable_systemd="no" ]) AM_CONDITIONAL(INSTALL_SYSTEMD, test x$enable_systemd = xyes) AC_ARG_ENABLE([upstart], [ --enable-upstart : Install upstart service files],, [ enable_upstart="no" ]) AM_CONDITIONAL(INSTALL_UPSTART, test x$enable_upstart = xyes) AC_ARG_WITH([initddir], [ --with-initddir=DIR : path to init script directory. ], [ INITDDIR="$withval" ], [ INITDDIR="$sysconfdir/init.d" ]) AC_ARG_WITH([systemddir], [ --with-systemddir=DIR : path to systemd unit files directory. ], [ SYSTEMDDIR="$withval" ], [ SYSTEMDDIR="/lib/systemd/system" ]) AC_ARG_WITH([upstartdir], [ --with-upstartdir=DIR : path to upstart config files directory. ], [ UPSTARTDIR="$withval" ], [ UPSTARTDIR="$sysconfdir/init" ]) AC_ARG_WITH([initwrappersdir], [ --with-initwrappersdir=DIR : path to init wrappers files directory. ], [ INITWRAPPERSDIR="$withval" ], [ INITWRAPPERSDIR="$datarootdir/corosync" ]) AC_ARG_ENABLE([snmp], [ --enable-snmp : SNMP protocol support ], [ default="no" ]) AC_ARG_ENABLE([xmlconf], [ --enable-xmlconf : XML configuration support ],, [ enable_xmlconf="no" ]) AM_CONDITIONAL(INSTALL_XMLCONF, test x$enable_xmlconf = xyes) AC_ARG_ENABLE([qdevices], [ --enable-qdevices : Quorum devices support ],, [ enable_qdevices="no" ]) AM_CONDITIONAL(BUILD_QDEVICES, test x$enable_qdevices = xyes) # *FLAGS handling goes here ENV_CFLAGS="$CFLAGS" ENV_CPPFLAGS="$CPPFLAGS" ENV_LDFLAGS="$LDFLAGS" # debug build stuff if test "x${enable_debug}" = xyes; then AC_DEFINE_UNQUOTED([DEBUG], [1], [Compiling Debugging code]) OPT_CFLAGS="-O0" PACKAGE_FEATURES="$PACKAGE_FEATURES debug" else OPT_CFLAGS="-O3" fi # gdb flags if test "x${GCC}" = xyes; then GDB_FLAGS="-ggdb3" else GDB_FLAGS="-g" fi # Look for dbus-1 if test "x${enable_dbus}" = xyes; then PKG_CHECK_MODULES([DBUS],[dbus-1]) AC_DEFINE_UNQUOTED([HAVE_DBUS], 1, [have dbus]) PACKAGE_FEATURES="$PACKAGE_FEATURES dbus" WITH_LIST="$WITH_LIST --with dbus" fi if test "x${enable_testagents}" = xyes; then AC_DEFINE_UNQUOTED([HAVE_TESTAGENTS], 1, [have testagents]) PACKAGE_FEATURES="$PACKAGE_FEATURES testagents" WITH_LIST="$WITH_LIST --with testagents" fi if test "x${enable_rdma}" = xyes; then PKG_CHECK_MODULES([rdmacm],[rdmacm]) PKG_CHECK_MODULES([ibverbs],[ibverbs]) AC_DEFINE_UNQUOTED([HAVE_RDMA], 1, [have rdmacm]) PACKAGE_FEATURES="$PACKAGE_FEATURES rdma" WITH_LIST="$WITH_LIST --with rdma" fi if test "x${enable_monitoring}" = xyes; then PKG_CHECK_MODULES([statgrab], [libstatgrab]) PKG_CHECK_MODULES([statgrabge090], [libstatgrab >= 0.90], AC_DEFINE_UNQUOTED([HAVE_LIBSTATGRAB_GE_090], 1, [have libstatgrab >= 0.90]), TMP_VARIABLE=1) AC_DEFINE_UNQUOTED([HAVE_MONITORING], 1, [have resource monitoring]) PACKAGE_FEATURES="$PACKAGE_FEATURES monitoring" WITH_LIST="$WITH_LIST --with monitoring" fi if test "x${enable_watchdog}" = xyes; then AC_CHECK_HEADER([linux/watchdog.h], [], [AC_MSG_ERROR([watchdog requires linux/watchdog.h])]) AC_CHECK_HEADER([linux/reboot.h], [], [AC_MSG_ERROR([watchdog requires linux/reboot.h])]) AC_DEFINE_UNQUOTED([HAVE_WATCHDOG], 1, [have watchdog]) PACKAGE_FEATURES="$PACKAGE_FEATURES watchdog" WITH_LIST="$WITH_LIST --with watchdog" fi if test "x${enable_augeas}" = xyes; then PACKAGE_FEATURES="$PACKAGE_FEATURES augeas" fi if test "x${enable_systemd}" = xyes; then PACKAGE_FEATURES="$PACKAGE_FEATURES systemd" WITH_LIST="$WITH_LIST --with systemd" fi if test "x${enable_upstart}" = xyes; then PACKAGE_FEATURES="$PACKAGE_FEATURES upstart" WITH_LIST="$WITH_LIST --with upstart" fi if test "x${enable_xmlconf}" = xyes; then PACKAGE_FEATURES="$PACKAGE_FEATURES xmlconf" WITH_LIST="$WITH_LIST --with xmlconf" fi if test "x${enable_qdevices}" = xyes; then PACKAGE_FEATURES="$PACKAGE_FEATURES qdevices" fi do_snmp=0 if test "x${enable_snmp}" = xyes; then AC_PATH_PROGS([SNMPCONFIG], [net-snmp-config]) if test "x${SNMPCONFIG}" != "x"; then AC_MSG_CHECKING([for snmp includes]) SNMP_PREFIX=`$SNMPCONFIG --prefix` SNMP_INCLUDES="-I$SNMP_PREFIX/include" AC_MSG_RESULT([$SNMP_INCLUDES]) AC_MSG_CHECKING([for snmp libraries]) SNMP_LIBS=`$SNMPCONFIG --libs` AC_MSG_RESULT([$SNMP_LIBS]) AC_SUBST([SNMP_LIBS]) saveCFLAGS="$CFLAGS" CFLAGS="$CFLAGS $SNMP_INCLUDES" AC_CHECK_HEADERS([net-snmp/net-snmp-config.h]) CFLAGS="$saveCFLAGS" if test "x${ac_cv_header_net_snmp_net_snmp_config_h}" != "xyes"; then AC_MSG_ERROR([Unable to use net-snmp/net-snmp-config.h]) fi savedLibs=$LIBS LIBS="$LIBS $SNMP_LIBS" AC_CHECK_FUNCS([netsnmp_transport_open_client]) if test $ac_cv_func_netsnmp_transport_open_client != yes; then AC_CHECK_FUNCS([netsnmp_tdomain_transport]) if test $ac_cv_func_netsnmp_tdomain_transport != yes; then AC_MSG_ERROR([No usable SNMP client transport implementation found]) fi else AC_DEFINE_UNQUOTED([NETSNMPV54], $NETSNMP_NEW_SUPPORT, [have net-snmp5.4 over]) fi LIBS=$savedLibs do_snmp=1 PACKAGE_FEATURES="$PACKAGE_FEATURES snmp" WITH_LIST="$WITH_LIST --with snmp" AC_DEFINE_UNQUOTED([ENABLE_SNMP], $do_snmp, [Build in support for sending SNMP traps]) else AC_MSG_ERROR([You need the net_snmp development package to continue.]) fi fi AM_CONDITIONAL(BUILD_SNMP, test "${do_snmp}" = "1") # extra warnings EXTRA_WARNINGS="" WARNLIST=" all shadow missing-prototypes missing-declarations strict-prototypes declaration-after-statement pointer-arith write-strings cast-align bad-function-cast missing-format-attribute format=2 format-security format-nonliteral no-long-long unsigned-char gnu89-inline no-strict-aliasing " for j in $WARNLIST; do if cc_supports_flag -W$j; then EXTRA_WARNINGS="$EXTRA_WARNINGS -W$j"; fi done if test "x${enable_coverage}" = xyes && \ cc_supports_flag -ftest-coverage && \ cc_supports_flag -fprofile-arcs ; then AC_MSG_NOTICE([Enabling Coverage (enable -O0 by default)]) OPT_CFLAGS="-O0" COVERAGE_CFLAGS="-ftest-coverage -fprofile-arcs" COVERAGE_LDFLAGS="-ftest-coverage -fprofile-arcs" PACKAGE_FEATURES="$PACKAGE_FEATURES coverage" else COVERAGE_CFLAGS="" COVERAGE_LDFLAGS="" fi if test "x${enable_small_memory_footprint}" = xyes ; then AC_DEFINE_UNQUOTED([HAVE_SMALL_MEMORY_FOOTPRINT], 1, [have small_memory_footprint]) PACKAGE_FEATURES="$PACKAGE_FEATURES small-memory-footprint" fi if test "x${enable_ansi}" = xyes && \ cc_supports_flag -std=iso9899:199409 ; then AC_MSG_NOTICE([Enabling ANSI Compatibility]) ANSI_CPPFLAGS="-ansi -DANSI_ONLY" PACKAGE_FEATURES="$PACKAGE_FEATURES ansi" else ANSI_CPPFLAGS="" fi if test "x${enable_fatal_warnings}" = xyes && \ cc_supports_flag -Werror ; then AC_MSG_NOTICE([Enabling Fatal Warnings (-Werror)]) WERROR_CFLAGS="-Werror" PACKAGE_FEATURES="$PACKAGE_FEATURES fatal-warnings" else WERROR_CFLAGS="" fi # don't add addtional cflags if test "x${enable_user_flags}" = xyes; then OPT_CFLAGS="" GDB_FLAGS="" EXTRA_WARNINGS="" fi if test "x${enable_secure_build}" = xyes; then # stolen from apache configure snippet AC_CACHE_CHECK([whether $CC accepts PIE flags], [ap_cv_cc_pie], [ save_CFLAGS=$CFLAGS save_LDFLAGS=$LDFLAGS CFLAGS="$CFLAGS -fPIE" LDFLAGS="$LDFLAGS -pie" AC_TRY_RUN([static int foo[30000]; int main () { return 0; }], [ap_cv_cc_pie=yes], [ap_cv_cc_pie=no], [ap_cv_cc_pie=yes]) CFLAGS=$save_CFLAGS LDFLAGS=$save_LDFLAGS ]) if test "$ap_cv_cc_pie" = "yes"; then SEC_FLAGS="$SEC_FLAGS -fPIE" SEC_LDFLAGS="$SEC_LDFLAGS -pie" PACKAGE_FEATURES="$PACKAGE_FEATURES pie" fi # similar to above AC_CACHE_CHECK([whether $CC accepts RELRO flags], [ap_cv_cc_relro], [ save_LDFLAGS=$LDFLAGS LDFLAGS="$LDFLAGS -Wl,-z,relro" AC_TRY_RUN([static int foo[30000]; int main () { return 0; }], [ap_cv_cc_relro=yes], [ap_cv_cc_relro=no], [ap_cv_cc_relro=yes]) LDFLAGS=$save_LDFLAGS ]) if test "$ap_cv_cc_relro" = "yes"; then SEC_LDFLAGS="$SEC_LDFLAGS -Wl,-z,relro" PACKAGE_FEATURES="$PACKAGE_FEATURES relro" fi AC_CACHE_CHECK([whether $CC accepts BINDNOW flags], [ap_cv_cc_bindnow], [ save_LDFLAGS=$LDFLAGS LDFLAGS="$LDFLAGS -Wl,-z,now" AC_TRY_RUN([static int foo[30000]; int main () { return 0; }], [ap_cv_cc_bindnow=yes], [ap_cv_cc_bindnow=no], [ap_cv_cc_bindnow=yes]) LDFLAGS=$save_LDFLAGS ]) if test "$ap_cv_cc_bindnow" = "yes"; then SEC_LDFLAGS="$SEC_LDFLAGS -Wl,-z,now" PACKAGE_FEATURES="$PACKAGE_FEATURES bindnow" fi fi AC_CACHE_CHECK([whether $CC accepts "--as-needed"], [ap_cv_cc_as_needed], [ save_LDFLAGS=$LDFLAGS LDFLAGS="$LDFLAGS -Wl,--as-needed" AC_TRY_RUN([static int foo[30000]; int main () { return 0; }], [ap_cv_cc_as_needed=yes], [ap_cv_cc_as_needed=no], [ap_cv_cc_as_needed=yes]) LDFLAGS=$save_LDFLAGS ]) AC_CACHE_CHECK([whether $CC accepts "--version-script"], [ap_cv_cc_version_script], [ save_LDFLAGS=$LDFLAGS LDFLAGS="$LDFLAGS -Wl,--version-script=conftest.versions" echo "CONFTEST { };" >conftest.versions AC_TRY_RUN([static int foo[30000]; int main () { return 0; }], [ap_cv_cc_version_script=yes], [ap_cv_cc_version_script=no], [ap_cv_cc_version_script=yes]) rm -f conftest.versions LDFLAGS=$save_LDFLAGS ]) if test "$ap_cv_cc_version_script" = "yes"; then AC_SUBST(VERSCRIPT_LDFLAGS, ["-Wl,--version-script=\$(srcdir)/lib\$(call get_libname,\$<).versions"]) else AC_SUBST(VERSCRIPT_LDFLAGS, [""]) fi # define global include dirs INCLUDE_DIRS="$INCLUDE_DIRS -I\$(top_builddir)/include -I\$(top_srcdir)/include" INCLUDE_DIRS="$INCLUDE_DIRS -I\$(top_builddir)/include/corosync -I\$(top_srcdir)/include/corosync" # final build of *FLAGS CFLAGS="$ENV_CFLAGS $lt_prog_compiler_pic $SEC_FLAGS $OPT_CFLAGS $GDB_FLAGS \ $COVERAGE_CFLAGS $EXTRA_WARNINGS \ $WERROR_CFLAGS $NSS_CFLAGS $LIBQB_CFLAGS \ $SNMP_INCLUDES" CPPFLAGS="$ENV_CPPFLAGS $ANSI_CPPFLAGS $INCLUDE_DIRS" LDFLAGS="$ENV_LDFLAGS $lt_prog_compiler_pic $SEC_LDFLAGS $COVERAGE_LDFLAGS" if test "$ap_cv_cc_as_needed" = "yes"; then LDFLAGS="$LDFLAGS -Wl,--as-needed" fi # substitute what we need: AC_SUBST([BASHPATH]) AC_SUBST([INITDDIR]) AC_SUBST([SYSTEMDDIR]) AC_SUBST([UPSTARTDIR]) INITWRAPPERSDIR=$(eval echo ${INITWRAPPERSDIR}) AC_SUBST([INITWRAPPERSDIR]) AC_SUBST([SOMAJOR]) AC_SUBST([SOMINOR]) AC_SUBST([SOMICRO]) AC_SUBST([SONAME]) AM_CONDITIONAL(INSTALL_TESTAGENTS, test "${enable_testagents}" = "yes") AM_CONDITIONAL(INSTALL_MIB, test "${do_snmp}" = "1") AM_CONDITIONAL(INSTALL_DBUSCONF, test "${enable_dbus}" = "yes") AM_CONDITIONAL(AUGTOOL, test -n "${AUGTOOL}") AC_SUBST([NSS_LDFLAGS]) AM_CONDITIONAL(BUILD_HTML_DOCS, test -n "${GROFF}") AC_SUBST([LINT_FLAGS]) AC_DEFINE_UNQUOTED([LOCALSTATEDIR], "$(eval echo ${localstatedir})", [localstate directory]) COROSYSCONFDIR=${sysconfdir}/corosync AC_SUBST([COROSYSCONFDIR]) AC_DEFINE_UNQUOTED([COROSYSCONFDIR], "$(eval echo ${COROSYSCONFDIR})", [corosync config directory]) AC_DEFINE_UNQUOTED([PACKAGE_FEATURES], "${PACKAGE_FEATURES}", [corosync built-in features]) AC_OUTPUT AC_MSG_RESULT([]) AC_MSG_RESULT([$PACKAGE configuration:]) AC_MSG_RESULT([ Version = ${VERSION}]) AC_MSG_RESULT([ Prefix = ${prefix}]) AC_MSG_RESULT([ Executables = ${sbindir}]) AC_MSG_RESULT([ Man pages = ${mandir}]) AC_MSG_RESULT([ Doc dir = ${docdir}]) AC_MSG_RESULT([ Libraries = ${libdir}]) AC_MSG_RESULT([ Header files = ${includedir}]) AC_MSG_RESULT([ Arch-independent files = ${datadir}]) AC_MSG_RESULT([ State information = ${localstatedir}]) AC_MSG_RESULT([ System configuration = ${sysconfdir}]) AC_MSG_RESULT([ System init.d directory = ${INITDDIR}]) AC_MSG_RESULT([ System systemd directory = ${SYSTEMDDIR}]) AC_MSG_RESULT([ System upstart directory = ${UPSTARTDIR}]) AC_MSG_RESULT([ System init wraps dir = ${INITWRAPPERSDIR}]) AC_MSG_RESULT([ corosync config dir = ${COROSYSCONFDIR}]) AC_MSG_RESULT([ Features =${PACKAGE_FEATURES}]) AC_MSG_RESULT([]) AC_MSG_RESULT([$PACKAGE build info:]) AC_MSG_RESULT([ Library SONAME = ${SONAME}]) LIB_MSG_RESULT(m4_shift(local_soname_list))dnl AC_MSG_RESULT([ Default optimization = ${OPT_CFLAGS}]) AC_MSG_RESULT([ Default debug options = ${GDB_CFLAGS}]) AC_MSG_RESULT([ Extra compiler warnings = ${EXTRA_WARNING}]) AC_MSG_RESULT([ Env. defined CFLAG = ${ENV_CFLAGS}]) AC_MSG_RESULT([ Env. defined CPPFLAGS = ${ENV_CPPFLAGS}]) AC_MSG_RESULT([ Env. defined LDFLAGS = ${ENV_LDFLAGS}]) AC_MSG_RESULT([ ANSI defined CPPFLAGS = ${ANSI_CPPFLAGS}]) AC_MSG_RESULT([ Coverage CFLAGS = ${COVERAGE_CFLAGS}]) AC_MSG_RESULT([ Coverage LDFLAGS = ${COVERAGE_LDFLAGS}]) AC_MSG_RESULT([ Fatal War. CFLAGS = ${WERROR_CFLAGS}]) AC_MSG_RESULT([ Final CFLAGS = ${CFLAGS}]) AC_MSG_RESULT([ Final CPPFLAGS = ${CPPFLAGS}]) AC_MSG_RESULT([ Final LDFLAGS = ${LDFLAGS}]) diff --git a/corosync.spec.in b/corosync.spec.in index 3ca75b75..a2ba584e 100644 --- a/corosync.spec.in +++ b/corosync.spec.in @@ -1,334 +1,335 @@ @ALPHATAG@ @NUMCOMM@ @DIRTY@ # Conditionals # Invoke "rpmbuild --without " or "rpmbuild --with " # to disable or enable specific features %bcond_with testagents %bcond_with watchdog %bcond_with monitoring %bcond_with snmp %bcond_with dbus %bcond_with rdma %bcond_with systemd %bcond_with upstart %bcond_with xmlconf %bcond_with runautogen %global gitver %{?numcomm:.%{numcomm}}%{?alphatag:.%{alphatag}}%{?dirty:.%{dirty}} %global gittarver %{?numcomm:.%{numcomm}}%{?alphatag:-%{alphatag}}%{?dirty:-%{dirty}} Name: corosync Summary: The Corosync Cluster Engine and Application Programming Interfaces Version: @version@ Release: 1%{?gitver}%{?dist} License: BSD Group: System Environment/Base URL: http://www.corosync.org/ Source0: https://github.com/downloads/corosync/corosync/%{name}-%{version}%{?gittarver}.tar.gz # Runtime bits Requires: corosynclib = %{version}-%{release} Requires(pre): /usr/sbin/useradd Requires(post): /sbin/chkconfig Requires(preun): /sbin/chkconfig Conflicts: openais <= 0.89, openais-devel <= 0.89 # Build bits BuildRequires: groff BuildRequires: libqb-devel BuildRequires: nss-devel +BuildRequires: zlib-devel %if %{with runautogen} BuildRequires: autoconf automake libtool %endif %if %{with monitoring} BuildRequires: libstatgrab-devel %endif %if %{with rdma} BuildRequires: libibverbs-devel librdmacm-devel %endif %if %{with snmp} BuildRequires: net-snmp-devel %endif %if %{with dbus} BuildRequires: dbus-devel %endif %if %{with systemd} BuildRequires: systemd-units Requires(post): systemd Requires(preun): systemd Requires(postun): systemd %endif %if %{with xmlconf} Requires: libxslt %endif BuildRoot: %(mktemp -ud %{_tmppath}/%{name}-%{version}-%{release}-XXXXXX) %prep %setup -q -n %{name}-%{version}%{?gittarver} %build %if %{with runautogen} ./autogen.sh %endif %if %{with rdma} export ibverbs_CFLAGS=-I/usr/include/infiniband \ export ibverbs_LIBS=-libverbs \ export rdmacm_CFLAGS=-I/usr/include/rdma \ export rdmacm_LIBS=-lrdmacm \ %endif %{configure} \ %if %{with testagents} --enable-testagents \ %endif %if %{with watchdog} --enable-watchdog \ %endif %if %{with monitoring} --enable-monitoring \ %endif %if %{with snmp} --enable-snmp \ %endif %if %{with dbus} --enable-dbus \ %endif %if %{with rdma} --enable-rdma \ %endif %if %{with systemd} --enable-systemd \ %endif %if %{with upstart} --enable-upstart \ %endif %if %{with xmlconf} --enable-xmlconf \ %endif --with-initddir=%{_initrddir} \ --with-systemddir=%{_unitdir} \ --with-upstartdir=%{_sysconfdir}/init make %{_smp_mflags} %install rm -rf %{buildroot} make install DESTDIR=%{buildroot} %if %{with dbus} mkdir -p -m 0700 %{buildroot}/%{_sysconfdir}/dbus-1/system.d install -m 644 %{_builddir}/%{name}-%{version}%{?gittarver}/conf/corosync-signals.conf %{buildroot}/%{_sysconfdir}/dbus-1/system.d/corosync-signals.conf %endif ## tree fixup # drop static libs rm -f %{buildroot}%{_libdir}/*.a rm -f %{buildroot}%{_libdir}/*.la # drop docs and html docs for now rm -rf %{buildroot}%{_docdir}/* # /etc/sysconfig/corosync-notifyd mkdir -p %{buildroot}%{_sysconfdir}/sysconfig install -m 644 tools/corosync-notifyd.sysconfig.example \ %{buildroot}%{_sysconfdir}/sysconfig/corosync-notifyd # /etc/sysconfig/corosync install -m 644 init/corosync.sysconfig.example \ %{buildroot}%{_sysconfdir}/sysconfig/corosync %clean rm -rf %{buildroot} %description This package contains the Corosync Cluster Engine Executive, several default APIs and libraries, default configuration files, and an init script. %post %if %{with systemd} && 0%{?systemd_post:1} %systemd_post corosync.service %else if [ $1 -eq 1 ]; then /sbin/chkconfig --add corosync || : fi %endif %preun %if %{with systemd} && 0%{?systemd_preun:1} %systemd_preun corosync.service %else if [ $1 -eq 0 ]; then /sbin/service corosync stop &>/dev/null || : /sbin/chkconfig --del corosync || : fi %endif %postun %if %{with systemd} && 0%{?systemd_postun:1} %systemd_postun %endif %files %defattr(-,root,root,-) %doc LICENSE SECURITY %{_sbindir}/corosync %{_sbindir}/corosync-keygen %{_sbindir}/corosync-cmapctl %{_sbindir}/corosync-cfgtool %{_sbindir}/corosync-cpgtool %{_sbindir}/corosync-quorumtool %{_sbindir}/corosync-notifyd %{_bindir}/corosync-blackbox %if %{with xmlconf} %{_bindir}/corosync-xmlproc %config(noreplace) %{_sysconfdir}/corosync/corosync.xml.example %dir %{_datadir}/corosync %{_datadir}/corosync/xml2conf.xsl %{_mandir}/man8/corosync-xmlproc.8* %{_mandir}/man5/corosync.xml.5* %endif %dir %{_sysconfdir}/corosync %dir %{_sysconfdir}/corosync/uidgid.d %config(noreplace) %{_sysconfdir}/corosync/corosync.conf.example %config(noreplace) %{_sysconfdir}/corosync/corosync.conf.example.udpu %config(noreplace) %{_sysconfdir}/sysconfig/corosync-notifyd %config(noreplace) %{_sysconfdir}/sysconfig/corosync %if %{with dbus} %{_sysconfdir}/dbus-1/system.d/corosync-signals.conf %endif %if %{with snmp} %{_datadir}/snmp/mibs/COROSYNC-MIB.txt %endif %if %{with systemd} %{_unitdir}/corosync.service %{_unitdir}/corosync-notifyd.service %dir %{_datadir}/corosync %{_datadir}/corosync/corosync %{_datadir}/corosync/corosync-notifyd %else %{_initrddir}/corosync %{_initrddir}/corosync-notifyd %endif %if %{with upstart} %{_sysconfdir}/init/corosync.conf %{_sysconfdir}/init/corosync-notifyd.conf %endif %dir %{_localstatedir}/lib/corosync %dir %{_localstatedir}/log/cluster %{_mandir}/man8/corosync_overview.8* %{_mandir}/man8/corosync.8* %{_mandir}/man8/corosync-blackbox.8* %{_mandir}/man8/corosync-cmapctl.8* %{_mandir}/man8/corosync-keygen.8* %{_mandir}/man8/corosync-cfgtool.8* %{_mandir}/man8/corosync-cpgtool.8* %{_mandir}/man8/corosync-notifyd.8* %{_mandir}/man8/corosync-quorumtool.8* %{_mandir}/man5/corosync.conf.5* %{_mandir}/man5/votequorum.5* %{_mandir}/man8/cmap_keys.8* # optional testagent rpm # %if %{with testagents} %package -n corosync-testagents Summary: The Corosync Cluster Engine Test Agents Group: Development/Libraries Requires: %{name} = %{version}-%{release} %description -n corosync-testagents This package contains corosync test agents. %files -n corosync-testagents %defattr(755,root,root,-) %{_datadir}/corosync/tests/mem_leak_test.sh %{_datadir}/corosync/tests/net_breaker.sh %{_datadir}/corosync/tests/cmap-dispatch-deadlock.sh %{_datadir}/corosync/tests/shm_leak_audit.sh %{_bindir}/cpg_test_agent %{_bindir}/sam_test_agent %{_bindir}/votequorum_test_agent %endif # library # %package -n corosynclib Summary: The Corosync Cluster Engine Libraries Group: System Environment/Libraries Requires: %{name} = %{version}-%{release} %description -n corosynclib This package contains corosync libraries. %files -n corosynclib %defattr(-,root,root,-) %doc LICENSE %{_libdir}/libcfg.so.* %{_libdir}/libcpg.so.* %{_libdir}/libcmap.so.* %{_libdir}/libtotem_pg.so.* %{_libdir}/libquorum.so.* %{_libdir}/libvotequorum.so.* %{_libdir}/libsam.so.* %{_libdir}/libcorosync_common.so.* %post -n corosynclib -p /sbin/ldconfig %postun -n corosynclib -p /sbin/ldconfig %package -n corosynclib-devel Summary: The Corosync Cluster Engine Development Kit Group: Development/Libraries Requires: corosynclib = %{version}-%{release} Requires: pkgconfig Provides: corosync-devel = %{version} Obsoletes: corosync-devel < 0.92-7 %description -n corosynclib-devel This package contains include files and man pages used to develop using The Corosync Cluster Engine APIs. %files -n corosynclib-devel %defattr(-,root,root,-) %doc LICENSE %dir %{_includedir}/corosync/ %{_includedir}/corosync/corodefs.h %{_includedir}/corosync/cfg.h %{_includedir}/corosync/cmap.h %{_includedir}/corosync/corotypes.h %{_includedir}/corosync/cpg.h %{_includedir}/corosync/hdb.h %{_includedir}/corosync/sam.h %{_includedir}/corosync/quorum.h %{_includedir}/corosync/votequorum.h %dir %{_includedir}/corosync/totem/ %{_includedir}/corosync/totem/totem.h %{_includedir}/corosync/totem/totemip.h %{_includedir}/corosync/totem/totempg.h %{_libdir}/libcfg.so %{_libdir}/libcpg.so %{_libdir}/libcmap.so %{_libdir}/libtotem_pg.so %{_libdir}/libquorum.so %{_libdir}/libvotequorum.so %{_libdir}/libsam.so %{_libdir}/libcorosync_common.so %{_libdir}/pkgconfig/*.pc %{_mandir}/man3/cpg_*3* %{_mandir}/man3/quorum_*3* %{_mandir}/man3/votequorum_*3* %{_mandir}/man3/sam_*3* %{_mandir}/man8/cpg_overview.8* %{_mandir}/man8/votequorum_overview.8* %{_mandir}/man8/sam_overview.8* %{_mandir}/man3/cmap_*3* %{_mandir}/man8/cmap_overview.8* %{_mandir}/man8/quorum_overview.8* %changelog * @date@ Autotools generated version - @version@-1-@numcomm@.@alphatag@.@dirty@ - Autotools generated version diff --git a/exec/cpg.c b/exec/cpg.c index 1c6fbb90..a18b8503 100644 --- a/exec/cpg.c +++ b/exec/cpg.c @@ -1,2261 +1,2437 @@ /* - * Copyright (c) 2006-2012 Red Hat, Inc. + * Copyright (c) 2006-2015 Red Hat, Inc. * * All rights reserved. * * Author: Christine Caulfield (ccaulfie@redhat.com) * Author: Jan Friesse (jfriesse@redhat.com) * * This software licensed under BSD license, the text of which follows: * * 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. * - Neither the name of the MontaVista Software, Inc. nor the names of its * contributors may 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 CONTIBUTORS "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 OWNER 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. */ #include #ifdef HAVE_ALLOCA_H #include #endif #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #ifndef MAP_ANONYMOUS #define MAP_ANONYMOUS MAP_ANON #endif #include "service.h" LOGSYS_DECLARE_SUBSYS ("CPG"); #define GROUP_HASH_SIZE 32 enum cpg_message_req_types { MESSAGE_REQ_EXEC_CPG_PROCJOIN = 0, MESSAGE_REQ_EXEC_CPG_PROCLEAVE = 1, MESSAGE_REQ_EXEC_CPG_JOINLIST = 2, MESSAGE_REQ_EXEC_CPG_MCAST = 3, MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD = 4, - MESSAGE_REQ_EXEC_CPG_DOWNLIST = 5 + MESSAGE_REQ_EXEC_CPG_DOWNLIST = 5, + MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST = 6, }; struct zcb_mapped { struct list_head list; void *addr; size_t size; }; /* * state` exec deliver * match group name, pid -> if matched deliver for YES: * XXX indicates impossible state * * join leave mcast * UNJOINED XXX XXX NO * LEAVE_STARTED XXX YES(unjoined_enter) YES * JOIN_STARTED YES(join_started_enter) XXX NO * JOIN_COMPLETED XXX NO YES * * join_started_enter * set JOIN_COMPLETED * add entry to process_info list * unjoined_enter * set UNJOINED * delete entry from process_info list * * * library accept join error codes * UNJOINED YES(CS_OK) set JOIN_STARTED * LEAVE_STARTED NO(CS_ERR_BUSY) * JOIN_STARTED NO(CS_ERR_EXIST) * JOIN_COMPlETED NO(CS_ERR_EXIST) * * library accept leave error codes * UNJOINED NO(CS_ERR_NOT_EXIST) * LEAVE_STARTED NO(CS_ERR_NOT_EXIST) * JOIN_STARTED NO(CS_ERR_BUSY) * JOIN_COMPLETED YES(CS_OK) set LEAVE_STARTED * * library accept mcast * UNJOINED NO(CS_ERR_NOT_EXIST) * LEAVE_STARTED NO(CS_ERR_NOT_EXIST) * JOIN_STARTED YES(CS_OK) * JOIN_COMPLETED YES(CS_OK) */ enum cpd_state { CPD_STATE_UNJOINED, CPD_STATE_LEAVE_STARTED, CPD_STATE_JOIN_STARTED, CPD_STATE_JOIN_COMPLETED }; enum cpg_sync_state { CPGSYNC_DOWNLIST, CPGSYNC_JOINLIST }; enum cpg_downlist_state_e { CPG_DOWNLIST_NONE, CPG_DOWNLIST_WAITING_FOR_MESSAGES, CPG_DOWNLIST_APPLYING, }; static enum cpg_downlist_state_e downlist_state; static struct list_head downlist_messages_head; static struct list_head joinlist_messages_head; struct cpg_pd { void *conn; mar_cpg_name_t group_name; uint32_t pid; enum cpd_state cpd_state; unsigned int flags; int initial_totem_conf_sent; + uint64_t transition_counter; /* These two are used when sending fragmented messages */ + uint64_t initial_transition_counter; struct list_head list; struct list_head iteration_instance_list_head; struct list_head zcb_mapped_list_head; }; struct cpg_iteration_instance { hdb_handle_t handle; struct list_head list; struct list_head items_list_head; /* List of process_info */ struct list_head *current_pointer; }; DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL); DECLARE_LIST_INIT(cpg_pd_list_head); static unsigned int my_member_list[PROCESSOR_COUNT_MAX]; static unsigned int my_member_list_entries; static unsigned int my_old_member_list[PROCESSOR_COUNT_MAX]; static unsigned int my_old_member_list_entries = 0; static struct corosync_api_v1 *api = NULL; static enum cpg_sync_state my_sync_state = CPGSYNC_DOWNLIST; static mar_cpg_ring_id_t last_sync_ring_id; struct process_info { unsigned int nodeid; uint32_t pid; mar_cpg_name_t group; struct list_head list; /* on the group_info members list */ }; DECLARE_LIST_INIT(process_info_list_head); struct join_list_entry { uint32_t pid; mar_cpg_name_t group_name; }; /* * Service Interfaces required by service_message_handler struct */ static char *cpg_exec_init_fn (struct corosync_api_v1 *); static int cpg_lib_init_fn (void *conn); static int cpg_lib_exit_fn (void *conn); static void message_handler_req_exec_cpg_procjoin ( const void *message, unsigned int nodeid); static void message_handler_req_exec_cpg_procleave ( const void *message, unsigned int nodeid); static void message_handler_req_exec_cpg_joinlist ( const void *message, unsigned int nodeid); static void message_handler_req_exec_cpg_mcast ( const void *message, unsigned int nodeid); +static void message_handler_req_exec_cpg_partial_mcast ( + const void *message, + unsigned int nodeid); + static void message_handler_req_exec_cpg_downlist_old ( const void *message, unsigned int nodeid); static void message_handler_req_exec_cpg_downlist ( const void *message, unsigned int nodeid); static void exec_cpg_procjoin_endian_convert (void *msg); static void exec_cpg_joinlist_endian_convert (void *msg); static void exec_cpg_mcast_endian_convert (void *msg); +static void exec_cpg_partial_mcast_endian_convert (void *msg); + static void exec_cpg_downlist_endian_convert_old (void *msg); static void exec_cpg_downlist_endian_convert (void *msg); static void message_handler_req_lib_cpg_join (void *conn, const void *message); static void message_handler_req_lib_cpg_leave (void *conn, const void *message); static void message_handler_req_lib_cpg_finalize (void *conn, const void *message); static void message_handler_req_lib_cpg_mcast (void *conn, const void *message); +static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message); + static void message_handler_req_lib_cpg_membership (void *conn, const void *message); static void message_handler_req_lib_cpg_local_get (void *conn, const void *message); static void message_handler_req_lib_cpg_iteration_initialize ( void *conn, const void *message); static void message_handler_req_lib_cpg_iteration_next ( void *conn, const void *message); static void message_handler_req_lib_cpg_iteration_finalize ( void *conn, const void *message); static void message_handler_req_lib_cpg_zc_alloc ( void *conn, const void *message); static void message_handler_req_lib_cpg_zc_free ( void *conn, const void *message); static void message_handler_req_lib_cpg_zc_execute ( void *conn, const void *message); static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason); static int cpg_exec_send_downlist(void); static int cpg_exec_send_joinlist(void); static void downlist_messages_delete (void); static void downlist_master_choose_and_send (void); static void joinlist_inform_clients (void); static void joinlist_messages_delete (void); static void cpg_sync_init ( const unsigned int *trans_list, size_t trans_list_entries, const unsigned int *member_list, size_t member_list_entries, const struct memb_ring_id *ring_id); static int cpg_sync_process (void); static void cpg_sync_activate (void); static void cpg_sync_abort (void); static void do_proc_join( const mar_cpg_name_t *name, uint32_t pid, unsigned int nodeid, int reason); static void do_proc_leave( const mar_cpg_name_t *name, uint32_t pid, unsigned int nodeid, int reason); static int notify_lib_totem_membership ( void *conn, int member_list_entries, const unsigned int *member_list); static inline int zcb_all_free ( struct cpg_pd *cpd); static char *cpg_print_group_name ( const mar_cpg_name_t *group); /* * Library Handler Definition */ static struct corosync_lib_handler cpg_lib_engine[] = { { /* 0 - MESSAGE_REQ_CPG_JOIN */ .lib_handler_fn = message_handler_req_lib_cpg_join, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 1 - MESSAGE_REQ_CPG_LEAVE */ .lib_handler_fn = message_handler_req_lib_cpg_leave, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 2 - MESSAGE_REQ_CPG_MCAST */ .lib_handler_fn = message_handler_req_lib_cpg_mcast, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 3 - MESSAGE_REQ_CPG_MEMBERSHIP */ .lib_handler_fn = message_handler_req_lib_cpg_membership, .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED }, { /* 4 - MESSAGE_REQ_CPG_LOCAL_GET */ .lib_handler_fn = message_handler_req_lib_cpg_local_get, .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED }, { /* 5 - MESSAGE_REQ_CPG_ITERATIONINITIALIZE */ .lib_handler_fn = message_handler_req_lib_cpg_iteration_initialize, .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED }, { /* 6 - MESSAGE_REQ_CPG_ITERATIONNEXT */ .lib_handler_fn = message_handler_req_lib_cpg_iteration_next, .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED }, { /* 7 - MESSAGE_REQ_CPG_ITERATIONFINALIZE */ .lib_handler_fn = message_handler_req_lib_cpg_iteration_finalize, .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED }, { /* 8 - MESSAGE_REQ_CPG_FINALIZE */ .lib_handler_fn = message_handler_req_lib_cpg_finalize, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 9 */ .lib_handler_fn = message_handler_req_lib_cpg_zc_alloc, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 10 */ .lib_handler_fn = message_handler_req_lib_cpg_zc_free, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, { /* 11 */ .lib_handler_fn = message_handler_req_lib_cpg_zc_execute, .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED }, - + { /* 12 */ + .lib_handler_fn = message_handler_req_lib_cpg_partial_mcast, + .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED + }, }; static struct corosync_exec_handler cpg_exec_engine[] = { { /* 0 - MESSAGE_REQ_EXEC_CPG_PROCJOIN */ .exec_handler_fn = message_handler_req_exec_cpg_procjoin, .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert }, { /* 1 - MESSAGE_REQ_EXEC_CPG_PROCLEAVE */ .exec_handler_fn = message_handler_req_exec_cpg_procleave, .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert }, { /* 2 - MESSAGE_REQ_EXEC_CPG_JOINLIST */ .exec_handler_fn = message_handler_req_exec_cpg_joinlist, .exec_endian_convert_fn = exec_cpg_joinlist_endian_convert }, { /* 3 - MESSAGE_REQ_EXEC_CPG_MCAST */ .exec_handler_fn = message_handler_req_exec_cpg_mcast, .exec_endian_convert_fn = exec_cpg_mcast_endian_convert }, { /* 4 - MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD */ .exec_handler_fn = message_handler_req_exec_cpg_downlist_old, .exec_endian_convert_fn = exec_cpg_downlist_endian_convert_old }, { /* 5 - MESSAGE_REQ_EXEC_CPG_DOWNLIST */ .exec_handler_fn = message_handler_req_exec_cpg_downlist, .exec_endian_convert_fn = exec_cpg_downlist_endian_convert }, + { /* 6 - MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST */ + .exec_handler_fn = message_handler_req_exec_cpg_partial_mcast, + .exec_endian_convert_fn = exec_cpg_partial_mcast_endian_convert + }, }; struct corosync_service_engine cpg_service_engine = { .name = "corosync cluster closed process group service v1.01", .id = CPG_SERVICE, .priority = 1, .private_data_size = sizeof (struct cpg_pd), .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED, .allow_inquorate = CS_LIB_ALLOW_INQUORATE, .lib_init_fn = cpg_lib_init_fn, .lib_exit_fn = cpg_lib_exit_fn, .lib_engine = cpg_lib_engine, .lib_engine_count = sizeof (cpg_lib_engine) / sizeof (struct corosync_lib_handler), .exec_init_fn = cpg_exec_init_fn, .exec_dump_fn = NULL, .exec_engine = cpg_exec_engine, .exec_engine_count = sizeof (cpg_exec_engine) / sizeof (struct corosync_exec_handler), .sync_init = cpg_sync_init, .sync_process = cpg_sync_process, .sync_activate = cpg_sync_activate, .sync_abort = cpg_sync_abort }; struct corosync_service_engine *cpg_get_service_engine_ver0 (void) { return (&cpg_service_engine); } struct req_exec_cpg_procjoin { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); mar_uint32_t reason __attribute__((aligned(8))); }; struct req_exec_cpg_mcast { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t msglen __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); mar_message_source_t source __attribute__((aligned(8))); mar_uint8_t message[] __attribute__((aligned(8))); }; +struct req_exec_cpg_partial_mcast { + struct qb_ipc_request_header header __attribute__((aligned(8))); + mar_cpg_name_t group_name __attribute__((aligned(8))); + mar_uint32_t msglen __attribute__((aligned(8))); + mar_uint32_t fraglen __attribute__((aligned(8))); + mar_uint32_t pid __attribute__((aligned(8))); + mar_uint32_t type __attribute__((aligned(8))); + mar_message_source_t source __attribute__((aligned(8))); + mar_uint8_t message[] __attribute__((aligned(8))); +}; + struct req_exec_cpg_downlist_old { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_uint32_t left_nodes __attribute__((aligned(8))); mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8))); }; struct req_exec_cpg_downlist { struct qb_ipc_request_header header __attribute__((aligned(8))); /* merge decisions */ mar_uint32_t old_members __attribute__((aligned(8))); /* downlist below */ mar_uint32_t left_nodes __attribute__((aligned(8))); mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8))); }; struct downlist_msg { mar_uint32_t sender_nodeid; mar_uint32_t old_members __attribute__((aligned(8))); mar_uint32_t left_nodes __attribute__((aligned(8))); mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8))); struct list_head list; }; struct joinlist_msg { mar_uint32_t sender_nodeid; uint32_t pid; mar_cpg_name_t group_name; struct list_head list; }; static struct req_exec_cpg_downlist g_req_exec_cpg_downlist; /* * Function print group name. It's not reentrant */ static char *cpg_print_group_name(const mar_cpg_name_t *group) { static char res[CPG_MAX_NAME_LENGTH * 4 + 1]; int dest_pos = 0; char c; int i; for (i = 0; i < group->length; i++) { c = group->value[i]; if (c >= ' ' && c < 0x7f && c != '\\') { res[dest_pos++] = c; } else { if (c == '\\') { res[dest_pos++] = '\\'; res[dest_pos++] = '\\'; } else { snprintf(res + dest_pos, sizeof(res) - dest_pos, "\\x%02X", c); dest_pos += 4; } } } res[dest_pos] = 0; return (res); } static void cpg_sync_init ( const unsigned int *trans_list, size_t trans_list_entries, const unsigned int *member_list, size_t member_list_entries, const struct memb_ring_id *ring_id) { int entries; int i, j; int found; my_sync_state = CPGSYNC_DOWNLIST; memcpy (my_member_list, member_list, member_list_entries * sizeof (unsigned int)); my_member_list_entries = member_list_entries; last_sync_ring_id.nodeid = ring_id->rep.nodeid; last_sync_ring_id.seq = ring_id->seq; downlist_state = CPG_DOWNLIST_WAITING_FOR_MESSAGES; entries = 0; /* * Determine list of nodeids for downlist message */ for (i = 0; i < my_old_member_list_entries; i++) { found = 0; for (j = 0; j < trans_list_entries; j++) { if (my_old_member_list[i] == trans_list[j]) { found = 1; break; } } if (found == 0) { g_req_exec_cpg_downlist.nodeids[entries++] = my_old_member_list[i]; } } g_req_exec_cpg_downlist.left_nodes = entries; } static int cpg_sync_process (void) { int res = -1; if (my_sync_state == CPGSYNC_DOWNLIST) { res = cpg_exec_send_downlist(); if (res == -1) { return (-1); } my_sync_state = CPGSYNC_JOINLIST; } if (my_sync_state == CPGSYNC_JOINLIST) { res = cpg_exec_send_joinlist(); } return (res); } static void cpg_sync_activate (void) { memcpy (my_old_member_list, my_member_list, my_member_list_entries * sizeof (unsigned int)); my_old_member_list_entries = my_member_list_entries; if (downlist_state == CPG_DOWNLIST_WAITING_FOR_MESSAGES) { downlist_master_choose_and_send (); } joinlist_inform_clients (); downlist_messages_delete (); downlist_state = CPG_DOWNLIST_NONE; joinlist_messages_delete (); notify_lib_totem_membership (NULL, my_member_list_entries, my_member_list); } static void cpg_sync_abort (void) { downlist_state = CPG_DOWNLIST_NONE; downlist_messages_delete (); joinlist_messages_delete (); } static int notify_lib_totem_membership ( void *conn, int member_list_entries, const unsigned int *member_list) { struct list_head *iter; char *buf; int size; struct res_lib_cpg_totem_confchg_callback *res; size = sizeof(struct res_lib_cpg_totem_confchg_callback) + sizeof(mar_uint32_t) * (member_list_entries); buf = alloca(size); if (!buf) return CS_ERR_LIBRARY; res = (struct res_lib_cpg_totem_confchg_callback *)buf; res->member_list_entries = member_list_entries; res->header.size = size; res->header.id = MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK; res->header.error = CS_OK; memcpy (&res->ring_id, &last_sync_ring_id, sizeof (mar_cpg_ring_id_t)); memcpy (res->member_list, member_list, res->member_list_entries * sizeof (mar_uint32_t)); if (conn == NULL) { for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) { struct cpg_pd *cpg_pd = list_entry (iter, struct cpg_pd, list); api->ipc_dispatch_send (cpg_pd->conn, buf, size); } } else { api->ipc_dispatch_send (conn, buf, size); } return CS_OK; } static int notify_lib_joinlist( const mar_cpg_name_t *group_name, void *conn, int joined_list_entries, mar_cpg_address_t *joined_list, int left_list_entries, mar_cpg_address_t *left_list, int id) { int size; char *buf; struct list_head *iter; int count; struct res_lib_cpg_confchg_callback *res; mar_cpg_address_t *retgi; count = 0; for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); if (mar_name_compare (&pi->group, group_name) == 0) { int i; int founded = 0; for (i = 0; i < left_list_entries; i++) { if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) { founded++; } } if (!founded) count++; } } size = sizeof(struct res_lib_cpg_confchg_callback) + sizeof(mar_cpg_address_t) * (count + left_list_entries + joined_list_entries); buf = alloca(size); if (!buf) return CS_ERR_LIBRARY; res = (struct res_lib_cpg_confchg_callback *)buf; res->joined_list_entries = joined_list_entries; res->left_list_entries = left_list_entries; res->member_list_entries = count; retgi = res->member_list; res->header.size = size; res->header.id = id; res->header.error = CS_OK; memcpy(&res->group_name, group_name, sizeof(mar_cpg_name_t)); for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi=list_entry (iter, struct process_info, list); if (mar_name_compare (&pi->group, group_name) == 0) { int i; int founded = 0; for (i = 0;i < left_list_entries; i++) { if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) { founded++; } } if (!founded) { retgi->nodeid = pi->nodeid; retgi->pid = pi->pid; retgi++; } } } if (left_list_entries) { memcpy (retgi, left_list, left_list_entries * sizeof(mar_cpg_address_t)); retgi += left_list_entries; } if (joined_list_entries) { memcpy (retgi, joined_list, joined_list_entries * sizeof(mar_cpg_address_t)); retgi += joined_list_entries; } if (conn) { api->ipc_dispatch_send (conn, buf, size); } else { for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) { struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list); if (mar_name_compare (&cpd->group_name, group_name) == 0) { assert (joined_list_entries <= 1); if (joined_list_entries) { if (joined_list[0].pid == cpd->pid && joined_list[0].nodeid == api->totem_nodeid_get()) { cpd->cpd_state = CPD_STATE_JOIN_COMPLETED; } } if (cpd->cpd_state == CPD_STATE_JOIN_COMPLETED || cpd->cpd_state == CPD_STATE_LEAVE_STARTED) { api->ipc_dispatch_send (cpd->conn, buf, size); + cpd->transition_counter++; } if (left_list_entries) { if (left_list[0].pid == cpd->pid && left_list[0].nodeid == api->totem_nodeid_get() && left_list[0].reason == CONFCHG_CPG_REASON_LEAVE) { cpd->pid = 0; memset (&cpd->group_name, 0, sizeof(cpd->group_name)); cpd->cpd_state = CPD_STATE_UNJOINED; } } } } } /* * Traverse thru cpds and send totem membership for cpd, where it is not send yet */ for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) { struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list); if ((cpd->flags & CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF) && (cpd->initial_totem_conf_sent == 0)) { cpd->initial_totem_conf_sent = 1; notify_lib_totem_membership (cpd->conn, my_old_member_list_entries, my_old_member_list); } } return CS_OK; } static void downlist_log(const char *msg, struct downlist_msg* dl) { log_printf (LOG_DEBUG, "%s: sender %s; members(old:%d left:%d)", msg, api->totem_ifaces_print(dl->sender_nodeid), dl->old_members, dl->left_nodes); } static struct downlist_msg* downlist_master_choose (void) { struct downlist_msg *cmp; struct downlist_msg *best = NULL; struct list_head *iter; uint32_t cmp_members; uint32_t best_members; uint32_t i; int ignore_msg; for (iter = downlist_messages_head.next; iter != &downlist_messages_head; iter = iter->next) { cmp = list_entry(iter, struct downlist_msg, list); downlist_log("comparing", cmp); ignore_msg = 0; for (i = 0; i < cmp->left_nodes; i++) { if (cmp->nodeids[i] == api->totem_nodeid_get()) { log_printf (LOG_DEBUG, "Ignoring this entry because I'm in the left list\n"); ignore_msg = 1; break; } } if (ignore_msg) { continue ; } if (best == NULL) { best = cmp; continue; } best_members = best->old_members - best->left_nodes; cmp_members = cmp->old_members - cmp->left_nodes; if (cmp_members > best_members) { best = cmp; } else if (cmp_members == best_members) { if (cmp->old_members > best->old_members) { best = cmp; } else if (cmp->old_members == best->old_members) { if (cmp->sender_nodeid < best->sender_nodeid) { best = cmp; } } } } assert (best != NULL); return best; } static void downlist_master_choose_and_send (void) { struct downlist_msg *stored_msg; struct list_head *iter; struct process_info *left_pi; qb_map_t *group_map; struct cpg_name cpg_group; mar_cpg_name_t group; struct confchg_data{ struct cpg_name cpg_group; mar_cpg_address_t left_list[CPG_MEMBERS_MAX]; int left_list_entries; struct list_head list; } *pcd; qb_map_iter_t *miter; int i, size; downlist_state = CPG_DOWNLIST_APPLYING; stored_msg = downlist_master_choose (); if (!stored_msg) { log_printf (LOGSYS_LEVEL_DEBUG, "NO chosen downlist"); return; } downlist_log("chosen downlist", stored_msg); group_map = qb_skiplist_create(); /* * only the cpg groups included in left nodes should receive * confchg event, so we will collect these cpg groups and * relative left_lists here. */ for (iter = process_info_list_head.next; iter != &process_info_list_head; ) { struct process_info *pi = list_entry(iter, struct process_info, list); iter = iter->next; left_pi = NULL; for (i = 0; i < stored_msg->left_nodes; i++) { if (pi->nodeid == stored_msg->nodeids[i]) { left_pi = pi; break; } } if (left_pi) { marshall_from_mar_cpg_name_t(&cpg_group, &left_pi->group); cpg_group.value[cpg_group.length] = 0; pcd = (struct confchg_data *)qb_map_get(group_map, cpg_group.value); if (pcd == NULL) { pcd = (struct confchg_data *)calloc(1, sizeof(struct confchg_data)); memcpy(&pcd->cpg_group, &cpg_group, sizeof(struct cpg_name)); qb_map_put(group_map, pcd->cpg_group.value, pcd); } size = pcd->left_list_entries; pcd->left_list[size].nodeid = left_pi->nodeid; pcd->left_list[size].pid = left_pi->pid; pcd->left_list[size].reason = CONFCHG_CPG_REASON_NODEDOWN; pcd->left_list_entries++; list_del (&left_pi->list); free (left_pi); } } /* send only one confchg event per cpg group */ miter = qb_map_iter_create(group_map); while (qb_map_iter_next(miter, (void **)&pcd)) { marshall_to_mar_cpg_name_t(&group, &pcd->cpg_group); log_printf (LOG_DEBUG, "left_list_entries:%d", pcd->left_list_entries); for (i=0; ileft_list_entries; i++) { log_printf (LOG_DEBUG, "left_list[%d] group:%s, ip:%s, pid:%d", i, cpg_print_group_name(&group), (char*)api->totem_ifaces_print(pcd->left_list[i].nodeid), pcd->left_list[i].pid); } /* send confchg event */ notify_lib_joinlist(&group, NULL, 0, NULL, pcd->left_list_entries, pcd->left_list, MESSAGE_RES_CPG_CONFCHG_CALLBACK); free(pcd); } qb_map_iter_free(miter); qb_map_destroy(group_map); } /* * Remove processes that might have left the group while we were suspended. */ static void joinlist_remove_zombie_pi_entries (void) { struct list_head *pi_iter; struct list_head *jl_iter; struct process_info *pi; struct joinlist_msg *stored_msg; int found; for (pi_iter = process_info_list_head.next; pi_iter != &process_info_list_head; ) { pi = list_entry (pi_iter, struct process_info, list); pi_iter = pi_iter->next; /* * Ignore local node */ if (pi->nodeid == api->totem_nodeid_get()) { continue ; } /* * Try to find message in joinlist messages */ found = 0; for (jl_iter = joinlist_messages_head.next; jl_iter != &joinlist_messages_head; jl_iter = jl_iter->next) { stored_msg = list_entry(jl_iter, struct joinlist_msg, list); if (stored_msg->sender_nodeid == api->totem_nodeid_get()) { continue ; } if (pi->nodeid == stored_msg->sender_nodeid && pi->pid == stored_msg->pid && mar_name_compare (&pi->group, &stored_msg->group_name) == 0) { found = 1; break ; } } if (!found) { do_proc_leave(&pi->group, pi->pid, pi->nodeid, CONFCHG_CPG_REASON_PROCDOWN); } } } static void joinlist_inform_clients (void) { struct joinlist_msg *stored_msg; struct list_head *iter; unsigned int i; i = 0; for (iter = joinlist_messages_head.next; iter != &joinlist_messages_head; iter = iter->next) { stored_msg = list_entry(iter, struct joinlist_msg, list); log_printf (LOG_DEBUG, "joinlist_messages[%u] group:%s, ip:%s, pid:%d", i++, cpg_print_group_name(&stored_msg->group_name), (char*)api->totem_ifaces_print(stored_msg->sender_nodeid), stored_msg->pid); /* Ignore our own messages */ if (stored_msg->sender_nodeid == api->totem_nodeid_get()) { continue ; } do_proc_join (&stored_msg->group_name, stored_msg->pid, stored_msg->sender_nodeid, CONFCHG_CPG_REASON_NODEUP); } joinlist_remove_zombie_pi_entries (); } static void downlist_messages_delete (void) { struct downlist_msg *stored_msg; struct list_head *iter, *iter_next; for (iter = downlist_messages_head.next; iter != &downlist_messages_head; iter = iter_next) { iter_next = iter->next; stored_msg = list_entry(iter, struct downlist_msg, list); list_del (&stored_msg->list); free (stored_msg); } } static void joinlist_messages_delete (void) { struct joinlist_msg *stored_msg; struct list_head *iter, *iter_next; for (iter = joinlist_messages_head.next; iter != &joinlist_messages_head; iter = iter_next) { iter_next = iter->next; stored_msg = list_entry(iter, struct joinlist_msg, list); list_del (&stored_msg->list); free (stored_msg); } list_init (&joinlist_messages_head); } static char *cpg_exec_init_fn (struct corosync_api_v1 *corosync_api) { list_init (&downlist_messages_head); list_init (&joinlist_messages_head); api = corosync_api; return (NULL); } static void cpg_iteration_instance_finalize (struct cpg_iteration_instance *cpg_iteration_instance) { struct list_head *iter, *iter_next; struct process_info *pi; for (iter = cpg_iteration_instance->items_list_head.next; iter != &cpg_iteration_instance->items_list_head; iter = iter_next) { iter_next = iter->next; pi = list_entry (iter, struct process_info, list); list_del (&pi->list); free (pi); } list_del (&cpg_iteration_instance->list); hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle); } static void cpg_pd_finalize (struct cpg_pd *cpd) { struct list_head *iter, *iter_next; struct cpg_iteration_instance *cpii; zcb_all_free(cpd); for (iter = cpd->iteration_instance_list_head.next; iter != &cpd->iteration_instance_list_head; iter = iter_next) { iter_next = iter->next; cpii = list_entry (iter, struct cpg_iteration_instance, list); cpg_iteration_instance_finalize (cpii); } list_del (&cpd->list); } static int cpg_lib_exit_fn (void *conn) { struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); log_printf(LOGSYS_LEVEL_DEBUG, "exit_fn for conn=%p", conn); if (cpd->group_name.length > 0 && cpd->cpd_state != CPD_STATE_LEAVE_STARTED) { cpg_node_joinleave_send (cpd->pid, &cpd->group_name, MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_PROCDOWN); } cpg_pd_finalize (cpd); api->ipc_refcnt_dec (conn); return (0); } static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason) { struct req_exec_cpg_procjoin req_exec_cpg_procjoin; struct iovec req_exec_cpg_iovec; int result; memcpy(&req_exec_cpg_procjoin.group_name, group_name, sizeof(mar_cpg_name_t)); req_exec_cpg_procjoin.pid = pid; req_exec_cpg_procjoin.reason = reason; req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin); req_exec_cpg_procjoin.header.id = SERVICE_ID_MAKE(CPG_SERVICE, fn); req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin; req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin); result = api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED); return (result); } /* Can byteswap join & leave messages */ static void exec_cpg_procjoin_endian_convert (void *msg) { struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = msg; req_exec_cpg_procjoin->pid = swab32(req_exec_cpg_procjoin->pid); swab_mar_cpg_name_t (&req_exec_cpg_procjoin->group_name); req_exec_cpg_procjoin->reason = swab32(req_exec_cpg_procjoin->reason); } static void exec_cpg_joinlist_endian_convert (void *msg_v) { char *msg = msg_v; struct qb_ipc_response_header *res = (struct qb_ipc_response_header *)msg; struct join_list_entry *jle = (struct join_list_entry *)(msg + sizeof(struct qb_ipc_response_header)); swab_mar_int32_t (&res->size); while ((const char*)jle < msg + res->size) { jle->pid = swab32(jle->pid); swab_mar_cpg_name_t (&jle->group_name); jle++; } } static void exec_cpg_downlist_endian_convert_old (void *msg) { } static void exec_cpg_downlist_endian_convert (void *msg) { struct req_exec_cpg_downlist *req_exec_cpg_downlist = msg; unsigned int i; req_exec_cpg_downlist->left_nodes = swab32(req_exec_cpg_downlist->left_nodes); req_exec_cpg_downlist->old_members = swab32(req_exec_cpg_downlist->old_members); for (i = 0; i < req_exec_cpg_downlist->left_nodes; i++) { req_exec_cpg_downlist->nodeids[i] = swab32(req_exec_cpg_downlist->nodeids[i]); } } static void exec_cpg_mcast_endian_convert (void *msg) { struct req_exec_cpg_mcast *req_exec_cpg_mcast = msg; swab_coroipc_request_header_t (&req_exec_cpg_mcast->header); swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name); req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid); req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen); swab_mar_message_source_t (&req_exec_cpg_mcast->source); } +static void exec_cpg_partial_mcast_endian_convert (void *msg) +{ + struct req_exec_cpg_partial_mcast *req_exec_cpg_mcast = msg; + + swab_coroipc_request_header_t (&req_exec_cpg_mcast->header); + swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name); + req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid); + req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen); + req_exec_cpg_mcast->fraglen = swab32(req_exec_cpg_mcast->fraglen); + req_exec_cpg_mcast->type = swab32(req_exec_cpg_mcast->type); + swab_mar_message_source_t (&req_exec_cpg_mcast->source); +} + static struct process_info *process_info_find(const mar_cpg_name_t *group_name, uint32_t pid, unsigned int nodeid) { struct list_head *iter; for (iter = process_info_list_head.next; iter != &process_info_list_head; ) { struct process_info *pi = list_entry (iter, struct process_info, list); iter = iter->next; if (pi->pid == pid && pi->nodeid == nodeid && mar_name_compare (&pi->group, group_name) == 0) { return pi; } } return NULL; } static void do_proc_join( const mar_cpg_name_t *name, uint32_t pid, unsigned int nodeid, int reason) { struct process_info *pi; struct process_info *pi_entry; mar_cpg_address_t notify_info; struct list_head *list; struct list_head *list_to_add = NULL; if (process_info_find (name, pid, nodeid) != NULL) { return ; } pi = malloc (sizeof (struct process_info)); if (!pi) { log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct"); return; } pi->nodeid = nodeid; pi->pid = pid; memcpy(&pi->group, name, sizeof(*name)); list_init(&pi->list); /* * Insert new process in sorted order so synchronization works properly */ list_to_add = &process_info_list_head; for (list = process_info_list_head.next; list != &process_info_list_head; list = list->next) { pi_entry = list_entry(list, struct process_info, list); if (pi_entry->nodeid > pi->nodeid || (pi_entry->nodeid == pi->nodeid && pi_entry->pid > pi->pid)) { break; } list_to_add = list; } list_add (&pi->list, list_to_add); notify_info.pid = pi->pid; notify_info.nodeid = nodeid; notify_info.reason = reason; notify_lib_joinlist(&pi->group, NULL, 1, ¬ify_info, 0, NULL, MESSAGE_RES_CPG_CONFCHG_CALLBACK); } static void do_proc_leave( const mar_cpg_name_t *name, uint32_t pid, unsigned int nodeid, int reason) { struct process_info *pi; struct list_head *iter; mar_cpg_address_t notify_info; notify_info.pid = pid; notify_info.nodeid = nodeid; notify_info.reason = reason; notify_lib_joinlist(name, NULL, 0, NULL, 1, ¬ify_info, MESSAGE_RES_CPG_CONFCHG_CALLBACK); for (iter = process_info_list_head.next; iter != &process_info_list_head; ) { pi = list_entry(iter, struct process_info, list); iter = iter->next; if (pi->pid == pid && pi->nodeid == nodeid && mar_name_compare (&pi->group, name)==0) { list_del (&pi->list); free (pi); } } } static void message_handler_req_exec_cpg_downlist_old ( const void *message, unsigned int nodeid) { log_printf (LOGSYS_LEVEL_WARNING, "downlist OLD from node 0x%x", nodeid); } static void message_handler_req_exec_cpg_downlist( const void *message, unsigned int nodeid) { const struct req_exec_cpg_downlist *req_exec_cpg_downlist = message; int i; struct list_head *iter; struct downlist_msg *stored_msg; int found; if (downlist_state != CPG_DOWNLIST_WAITING_FOR_MESSAGES) { log_printf (LOGSYS_LEVEL_WARNING, "downlist left_list: %d received in state %d", req_exec_cpg_downlist->left_nodes, downlist_state); return; } stored_msg = malloc (sizeof (struct downlist_msg)); stored_msg->sender_nodeid = nodeid; stored_msg->old_members = req_exec_cpg_downlist->old_members; stored_msg->left_nodes = req_exec_cpg_downlist->left_nodes; memcpy (stored_msg->nodeids, req_exec_cpg_downlist->nodeids, req_exec_cpg_downlist->left_nodes * sizeof (mar_uint32_t)); list_init (&stored_msg->list); list_add (&stored_msg->list, &downlist_messages_head); for (i = 0; i < my_member_list_entries; i++) { found = 0; for (iter = downlist_messages_head.next; iter != &downlist_messages_head; iter = iter->next) { stored_msg = list_entry(iter, struct downlist_msg, list); if (my_member_list[i] == stored_msg->sender_nodeid) { found = 1; } } if (!found) { return; } } downlist_master_choose_and_send (); } static void message_handler_req_exec_cpg_procjoin ( const void *message, unsigned int nodeid) { const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message; log_printf(LOGSYS_LEVEL_DEBUG, "got procjoin message from cluster node 0x%x (%s) for pid %u", nodeid, api->totem_ifaces_print(nodeid), (unsigned int)req_exec_cpg_procjoin->pid); do_proc_join (&req_exec_cpg_procjoin->group_name, req_exec_cpg_procjoin->pid, nodeid, CONFCHG_CPG_REASON_JOIN); } static void message_handler_req_exec_cpg_procleave ( const void *message, unsigned int nodeid) { const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message; log_printf(LOGSYS_LEVEL_DEBUG, "got procleave message from cluster node 0x%x (%s) for pid %u", nodeid, api->totem_ifaces_print(nodeid), (unsigned int)req_exec_cpg_procjoin->pid); do_proc_leave (&req_exec_cpg_procjoin->group_name, req_exec_cpg_procjoin->pid, nodeid, req_exec_cpg_procjoin->reason); } /* Got a proclist from another node */ static void message_handler_req_exec_cpg_joinlist ( const void *message_v, unsigned int nodeid) { const char *message = message_v; const struct qb_ipc_response_header *res = (const struct qb_ipc_response_header *)message; const struct join_list_entry *jle = (const struct join_list_entry *)(message + sizeof(struct qb_ipc_response_header)); struct joinlist_msg *stored_msg; log_printf(LOGSYS_LEVEL_DEBUG, "got joinlist message from node 0x%x", nodeid); while ((const char*)jle < message + res->size) { stored_msg = malloc (sizeof (struct joinlist_msg)); memset(stored_msg, 0, sizeof (struct joinlist_msg)); stored_msg->sender_nodeid = nodeid; stored_msg->pid = jle->pid; memcpy(&stored_msg->group_name, &jle->group_name, sizeof(mar_cpg_name_t)); list_init (&stored_msg->list); list_add (&stored_msg->list, &joinlist_messages_head); jle++; } } static void message_handler_req_exec_cpg_mcast ( const void *message, unsigned int nodeid) { const struct req_exec_cpg_mcast *req_exec_cpg_mcast = message; struct res_lib_cpg_deliver_callback res_lib_cpg_mcast; int msglen = req_exec_cpg_mcast->msglen; struct list_head *iter, *pi_iter; struct cpg_pd *cpd; struct iovec iovec[2]; int known_node = 0; res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_DELIVER_CALLBACK; res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen; res_lib_cpg_mcast.msglen = msglen; res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid; res_lib_cpg_mcast.nodeid = nodeid; memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name, sizeof(mar_cpg_name_t)); iovec[0].iov_base = (void *)&res_lib_cpg_mcast; iovec[0].iov_len = sizeof (res_lib_cpg_mcast); iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast); iovec[1].iov_len = msglen; for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; ) { cpd = list_entry(iter, struct cpg_pd, list); iter = iter->next; if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED || cpd->cpd_state == CPD_STATE_JOIN_COMPLETED) && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) { if (!known_node) { /* Try to find, if we know the node */ for (pi_iter = process_info_list_head.next; pi_iter != &process_info_list_head; pi_iter = pi_iter->next) { struct process_info *pi = list_entry (pi_iter, struct process_info, list); if (pi->nodeid == nodeid && mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) { known_node = 1; break; } } } if (!known_node) { log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message"); return ; } api->ipc_dispatch_iov_send (cpd->conn, iovec, 2); } } } +static void message_handler_req_exec_cpg_partial_mcast ( + const void *message, + unsigned int nodeid) +{ + const struct req_exec_cpg_partial_mcast *req_exec_cpg_mcast = message; + struct res_lib_cpg_partial_deliver_callback res_lib_cpg_mcast; + int msglen = req_exec_cpg_mcast->fraglen; + struct list_head *iter, *pi_iter; + struct cpg_pd *cpd; + struct iovec iovec[2]; + int known_node = 0; + + log_printf(LOGSYS_LEVEL_DEBUG, "Got fragmented message from node %d, size = %d bytes\n", nodeid, msglen); + + res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK; + res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen; + res_lib_cpg_mcast.fraglen = msglen; + res_lib_cpg_mcast.msglen = req_exec_cpg_mcast->msglen; + res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid; + res_lib_cpg_mcast.type = req_exec_cpg_mcast->type; + res_lib_cpg_mcast.nodeid = nodeid; + + memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name, + sizeof(mar_cpg_name_t)); + iovec[0].iov_base = (void *)&res_lib_cpg_mcast; + iovec[0].iov_len = sizeof (res_lib_cpg_mcast); + + iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast); + iovec[1].iov_len = msglen; + + for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; ) { + cpd = list_entry(iter, struct cpg_pd, list); + iter = iter->next; + + if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED || cpd->cpd_state == CPD_STATE_JOIN_COMPLETED) + && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) { + + if (!known_node) { + /* Try to find, if we know the node */ + for (pi_iter = process_info_list_head.next; + pi_iter != &process_info_list_head; pi_iter = pi_iter->next) { + + struct process_info *pi = list_entry (pi_iter, struct process_info, list); + + if (pi->nodeid == nodeid && + mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) { + known_node = 1; + break; + } + } + } + + if (!known_node) { + log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message"); + return ; + } + + api->ipc_dispatch_iov_send (cpd->conn, iovec, 2); + } + } +} + static int cpg_exec_send_downlist(void) { struct iovec iov; g_req_exec_cpg_downlist.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_DOWNLIST); g_req_exec_cpg_downlist.header.size = sizeof(struct req_exec_cpg_downlist); g_req_exec_cpg_downlist.old_members = my_old_member_list_entries; iov.iov_base = (void *)&g_req_exec_cpg_downlist; iov.iov_len = g_req_exec_cpg_downlist.header.size; return (api->totem_mcast (&iov, 1, TOTEM_AGREED)); } static int cpg_exec_send_joinlist(void) { int count = 0; struct list_head *iter; struct qb_ipc_response_header *res; char *buf; struct join_list_entry *jle; struct iovec req_exec_cpg_iovec; for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); if (pi->nodeid == api->totem_nodeid_get ()) { count++; } } /* Nothing to send */ if (!count) return 0; buf = alloca(sizeof(struct qb_ipc_response_header) + sizeof(struct join_list_entry) * count); if (!buf) { log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate joinlist buffer"); return -1; } jle = (struct join_list_entry *)(buf + sizeof(struct qb_ipc_response_header)); res = (struct qb_ipc_response_header *)buf; for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); if (pi->nodeid == api->totem_nodeid_get ()) { memcpy (&jle->group_name, &pi->group, sizeof (mar_cpg_name_t)); jle->pid = pi->pid; jle++; } } res->id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_JOINLIST); res->size = sizeof(struct qb_ipc_response_header)+sizeof(struct join_list_entry) * count; req_exec_cpg_iovec.iov_base = buf; req_exec_cpg_iovec.iov_len = res->size; return (api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED)); } static int cpg_lib_init_fn (void *conn) { struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); memset (cpd, 0, sizeof(struct cpg_pd)); cpd->conn = conn; list_add (&cpd->list, &cpg_pd_list_head); list_init (&cpd->iteration_instance_list_head); list_init (&cpd->zcb_mapped_list_head); api->ipc_refcnt_inc (conn); log_printf(LOGSYS_LEVEL_DEBUG, "lib_init_fn: conn=%p, cpd=%p", conn, cpd); return (0); } /* Join message from the library */ static void message_handler_req_lib_cpg_join (void *conn, const void *message) { const struct req_lib_cpg_join *req_lib_cpg_join = message; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); struct res_lib_cpg_join res_lib_cpg_join; cs_error_t error = CS_OK; struct list_head *iter; /* Test, if we don't have same pid and group name joined */ for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) { struct cpg_pd *cpd_item = list_entry (iter, struct cpg_pd, list); if (cpd_item->pid == req_lib_cpg_join->pid && mar_name_compare(&req_lib_cpg_join->group_name, &cpd_item->group_name) == 0) { /* We have same pid and group name joined -> return error */ error = CS_ERR_EXIST; goto response_send; } } /* * Same check must be done in process info list, because there may be not yet delivered * leave of client. */ for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); if (pi->nodeid == api->totem_nodeid_get () && pi->pid == req_lib_cpg_join->pid && mar_name_compare(&req_lib_cpg_join->group_name, &pi->group) == 0) { /* We have same pid and group name joined -> return error */ error = CS_ERR_TRY_AGAIN; goto response_send; } } if (req_lib_cpg_join->group_name.length > CPG_MAX_NAME_LENGTH) { error = CS_ERR_NAME_TOO_LONG; goto response_send; } switch (cpd->cpd_state) { case CPD_STATE_UNJOINED: error = CS_OK; cpd->cpd_state = CPD_STATE_JOIN_STARTED; cpd->pid = req_lib_cpg_join->pid; cpd->flags = req_lib_cpg_join->flags; memcpy (&cpd->group_name, &req_lib_cpg_join->group_name, sizeof (cpd->group_name)); cpg_node_joinleave_send (req_lib_cpg_join->pid, &req_lib_cpg_join->group_name, MESSAGE_REQ_EXEC_CPG_PROCJOIN, CONFCHG_CPG_REASON_JOIN); break; case CPD_STATE_LEAVE_STARTED: error = CS_ERR_BUSY; break; case CPD_STATE_JOIN_STARTED: error = CS_ERR_EXIST; break; case CPD_STATE_JOIN_COMPLETED: error = CS_ERR_EXIST; break; } response_send: res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join); res_lib_cpg_join.header.id = MESSAGE_RES_CPG_JOIN; res_lib_cpg_join.header.error = error; api->ipc_response_send (conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join)); } /* Leave message from the library */ static void message_handler_req_lib_cpg_leave (void *conn, const void *message) { struct res_lib_cpg_leave res_lib_cpg_leave; cs_error_t error = CS_OK; struct req_lib_cpg_leave *req_lib_cpg_leave = (struct req_lib_cpg_leave *)message; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); log_printf(LOGSYS_LEVEL_DEBUG, "got leave request on %p", conn); switch (cpd->cpd_state) { case CPD_STATE_UNJOINED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_LEAVE_STARTED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_JOIN_STARTED: error = CS_ERR_BUSY; break; case CPD_STATE_JOIN_COMPLETED: error = CS_OK; cpd->cpd_state = CPD_STATE_LEAVE_STARTED; cpg_node_joinleave_send (req_lib_cpg_leave->pid, &req_lib_cpg_leave->group_name, MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_LEAVE); break; } /* send return */ res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave); res_lib_cpg_leave.header.id = MESSAGE_RES_CPG_LEAVE; res_lib_cpg_leave.header.error = error; api->ipc_response_send(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave)); } /* Finalize message from library */ static void message_handler_req_lib_cpg_finalize ( void *conn, const void *message) { struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); struct res_lib_cpg_finalize res_lib_cpg_finalize; cs_error_t error = CS_OK; log_printf (LOGSYS_LEVEL_DEBUG, "cpg finalize for conn=%p", conn); /* * We will just remove cpd from list. After this call, connection will be * closed on lib side, and cpg_lib_exit_fn will be called */ list_del (&cpd->list); list_init (&cpd->list); res_lib_cpg_finalize.header.size = sizeof (res_lib_cpg_finalize); res_lib_cpg_finalize.header.id = MESSAGE_RES_CPG_FINALIZE; res_lib_cpg_finalize.header.error = error; api->ipc_response_send (conn, &res_lib_cpg_finalize, sizeof (res_lib_cpg_finalize)); } static int memory_map ( const char *path, size_t bytes, void **buf) { int32_t fd; void *addr; int32_t res; fd = open (path, O_RDWR, 0600); unlink (path); if (fd == -1) { return (-1); } res = ftruncate (fd, bytes); if (res == -1) { goto error_close_unlink; } addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); if (addr == MAP_FAILED) { goto error_close_unlink; } #ifdef MADV_NOSYNC madvise(addr, bytes, MADV_NOSYNC); #endif res = close (fd); if (res) { return (-1); } *buf = addr; return (0); error_close_unlink: close (fd); unlink(path); return -1; } static inline int zcb_alloc ( struct cpg_pd *cpd, const char *path_to_file, size_t size, void **addr) { struct zcb_mapped *zcb_mapped; unsigned int res; zcb_mapped = malloc (sizeof (struct zcb_mapped)); if (zcb_mapped == NULL) { return (-1); } res = memory_map ( path_to_file, size, addr); if (res == -1) { free (zcb_mapped); return (-1); } list_init (&zcb_mapped->list); zcb_mapped->addr = *addr; zcb_mapped->size = size; list_add_tail (&zcb_mapped->list, &cpd->zcb_mapped_list_head); return (0); } static inline int zcb_free (struct zcb_mapped *zcb_mapped) { unsigned int res; res = munmap (zcb_mapped->addr, zcb_mapped->size); list_del (&zcb_mapped->list); free (zcb_mapped); return (res); } static inline int zcb_by_addr_free (struct cpg_pd *cpd, void *addr) { struct list_head *list; struct zcb_mapped *zcb_mapped; unsigned int res = 0; for (list = cpd->zcb_mapped_list_head.next; list != &cpd->zcb_mapped_list_head; list = list->next) { zcb_mapped = list_entry (list, struct zcb_mapped, list); if (zcb_mapped->addr == addr) { res = zcb_free (zcb_mapped); break; } } return (res); } static inline int zcb_all_free ( struct cpg_pd *cpd) { struct list_head *list; struct zcb_mapped *zcb_mapped; for (list = cpd->zcb_mapped_list_head.next; list != &cpd->zcb_mapped_list_head;) { zcb_mapped = list_entry (list, struct zcb_mapped, list); list = list->next; zcb_free (zcb_mapped); } return (0); } union u { uint64_t server_addr; void *server_ptr; }; static uint64_t void2serveraddr (void *server_ptr) { union u u; u.server_ptr = server_ptr; return (u.server_addr); } static void *serveraddr2void (uint64_t server_addr) { union u u; u.server_addr = server_addr; return (u.server_ptr); }; static void message_handler_req_lib_cpg_zc_alloc ( void *conn, const void *message) { mar_req_coroipcc_zc_alloc_t *hdr = (mar_req_coroipcc_zc_alloc_t *)message; struct qb_ipc_response_header res_header; void *addr = NULL; struct coroipcs_zc_header *zc_header; unsigned int res; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); log_printf(LOGSYS_LEVEL_DEBUG, "path: %s", hdr->path_to_file); res = zcb_alloc (cpd, hdr->path_to_file, hdr->map_size, &addr); assert(res == 0); zc_header = (struct coroipcs_zc_header *)addr; zc_header->server_address = void2serveraddr(addr); res_header.size = sizeof (struct qb_ipc_response_header); res_header.id = 0; api->ipc_response_send (conn, &res_header, res_header.size); } static void message_handler_req_lib_cpg_zc_free ( void *conn, const void *message) { mar_req_coroipcc_zc_free_t *hdr = (mar_req_coroipcc_zc_free_t *)message; struct qb_ipc_response_header res_header; void *addr = NULL; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); log_printf(LOGSYS_LEVEL_DEBUG, " free'ing"); addr = serveraddr2void (hdr->server_address); zcb_by_addr_free (cpd, addr); res_header.size = sizeof (struct qb_ipc_response_header); res_header.id = 0; api->ipc_response_send ( conn, &res_header, res_header.size); } +/* Fragmented mcast message from the library */ +static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message) +{ + const struct req_lib_cpg_partial_mcast *req_lib_cpg_mcast = message; + struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); + mar_cpg_name_t group_name = cpd->group_name; + + struct iovec req_exec_cpg_iovec[2]; + struct req_exec_cpg_partial_mcast req_exec_cpg_mcast; + struct res_lib_cpg_partial_send res_lib_cpg_partial_send; + int msglen = req_lib_cpg_mcast->fraglen; + int result; + cs_error_t error = CS_ERR_NOT_EXIST; + + log_printf(LOGSYS_LEVEL_TRACE, "got fragmented mcast request on %p", conn); + log_printf(LOGSYS_LEVEL_DEBUG, "Sending fragmented message size = %d bytes\n", msglen); + + switch (cpd->cpd_state) { + case CPD_STATE_UNJOINED: + error = CS_ERR_NOT_EXIST; + break; + case CPD_STATE_LEAVE_STARTED: + error = CS_ERR_NOT_EXIST; + break; + case CPD_STATE_JOIN_STARTED: + error = CS_OK; + break; + case CPD_STATE_JOIN_COMPLETED: + error = CS_OK; + break; + } + + res_lib_cpg_partial_send.header.size = sizeof(res_lib_cpg_partial_send); + res_lib_cpg_partial_send.header.id = MESSAGE_RES_CPG_PARTIAL_SEND; + + if (req_lib_cpg_mcast->type == LIBCPG_PARTIAL_FIRST) { + cpd->initial_transition_counter = cpd->transition_counter; + } + if (cpd->transition_counter != cpd->initial_transition_counter) { + error = CS_ERR_INTERRUPT; + } + + if (error == CS_OK) { + req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen; + req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE, + MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST); + req_exec_cpg_mcast.pid = cpd->pid; + req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen; + req_exec_cpg_mcast.type = req_lib_cpg_mcast->type; + req_exec_cpg_mcast.fraglen = req_lib_cpg_mcast->fraglen; + api->ipc_source_set (&req_exec_cpg_mcast.source, conn); + memcpy(&req_exec_cpg_mcast.group_name, &group_name, + sizeof(mar_cpg_name_t)); + + req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast; + req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast); + req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message; + req_exec_cpg_iovec[1].iov_len = msglen; + + result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED); + assert(result == 0); + } else { + log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d", + conn, group_name.value, cpd->cpd_state, error); + } + + res_lib_cpg_partial_send.header.error = error; + api->ipc_response_send (conn, &res_lib_cpg_partial_send, + sizeof (res_lib_cpg_partial_send)); +} + /* Mcast message from the library */ static void message_handler_req_lib_cpg_mcast (void *conn, const void *message) { const struct req_lib_cpg_mcast *req_lib_cpg_mcast = message; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); mar_cpg_name_t group_name = cpd->group_name; struct iovec req_exec_cpg_iovec[2]; struct req_exec_cpg_mcast req_exec_cpg_mcast; int msglen = req_lib_cpg_mcast->msglen; int result; cs_error_t error = CS_ERR_NOT_EXIST; log_printf(LOGSYS_LEVEL_TRACE, "got mcast request on %p", conn); switch (cpd->cpd_state) { case CPD_STATE_UNJOINED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_LEAVE_STARTED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_JOIN_STARTED: error = CS_OK; break; case CPD_STATE_JOIN_COMPLETED: error = CS_OK; break; } if (error == CS_OK) { req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen; req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_MCAST); req_exec_cpg_mcast.pid = cpd->pid; req_exec_cpg_mcast.msglen = msglen; api->ipc_source_set (&req_exec_cpg_mcast.source, conn); memcpy(&req_exec_cpg_mcast.group_name, &group_name, sizeof(mar_cpg_name_t)); req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast; req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast); req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message; req_exec_cpg_iovec[1].iov_len = msglen; result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED); assert(result == 0); } else { log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d", conn, group_name.value, cpd->cpd_state, error); } } static void message_handler_req_lib_cpg_zc_execute ( void *conn, const void *message) { mar_req_coroipcc_zc_execute_t *hdr = (mar_req_coroipcc_zc_execute_t *)message; struct qb_ipc_request_header *header; struct res_lib_cpg_mcast res_lib_cpg_mcast; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); struct iovec req_exec_cpg_iovec[2]; struct req_exec_cpg_mcast req_exec_cpg_mcast; struct req_lib_cpg_mcast *req_lib_cpg_mcast; int result; cs_error_t error = CS_ERR_NOT_EXIST; log_printf(LOGSYS_LEVEL_TRACE, "got ZC mcast request on %p", conn); header = (struct qb_ipc_request_header *)(((char *)serveraddr2void(hdr->server_address) + sizeof (struct coroipcs_zc_header))); req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)header; switch (cpd->cpd_state) { case CPD_STATE_UNJOINED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_LEAVE_STARTED: error = CS_ERR_NOT_EXIST; break; case CPD_STATE_JOIN_STARTED: error = CS_OK; break; case CPD_STATE_JOIN_COMPLETED: error = CS_OK; break; } res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast); res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST; if (error == CS_OK) { req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + req_lib_cpg_mcast->msglen; req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_MCAST); req_exec_cpg_mcast.pid = cpd->pid; req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen; api->ipc_source_set (&req_exec_cpg_mcast.source, conn); memcpy(&req_exec_cpg_mcast.group_name, &cpd->group_name, sizeof(mar_cpg_name_t)); req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast; req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast); req_exec_cpg_iovec[1].iov_base = (char *)header + sizeof(struct req_lib_cpg_mcast); req_exec_cpg_iovec[1].iov_len = req_exec_cpg_mcast.msglen; result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED); if (result == 0) { res_lib_cpg_mcast.header.error = CS_OK; } else { res_lib_cpg_mcast.header.error = CS_ERR_TRY_AGAIN; } } else { res_lib_cpg_mcast.header.error = error; } api->ipc_response_send (conn, &res_lib_cpg_mcast, sizeof (res_lib_cpg_mcast)); } static void message_handler_req_lib_cpg_membership (void *conn, const void *message) { struct req_lib_cpg_membership_get *req_lib_cpg_membership_get = (struct req_lib_cpg_membership_get *)message; struct res_lib_cpg_membership_get res_lib_cpg_membership_get; struct list_head *iter; int member_count = 0; res_lib_cpg_membership_get.header.id = MESSAGE_RES_CPG_MEMBERSHIP; res_lib_cpg_membership_get.header.error = CS_OK; res_lib_cpg_membership_get.header.size = sizeof (struct res_lib_cpg_membership_get); for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); if (mar_name_compare (&pi->group, &req_lib_cpg_membership_get->group_name) == 0) { res_lib_cpg_membership_get.member_list[member_count].nodeid = pi->nodeid; res_lib_cpg_membership_get.member_list[member_count].pid = pi->pid; member_count += 1; } } res_lib_cpg_membership_get.member_count = member_count; api->ipc_response_send (conn, &res_lib_cpg_membership_get, sizeof (res_lib_cpg_membership_get)); } static void message_handler_req_lib_cpg_local_get (void *conn, const void *message) { struct res_lib_cpg_local_get res_lib_cpg_local_get; res_lib_cpg_local_get.header.size = sizeof (res_lib_cpg_local_get); res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET; res_lib_cpg_local_get.header.error = CS_OK; res_lib_cpg_local_get.local_nodeid = api->totem_nodeid_get (); api->ipc_response_send (conn, &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get)); } static void message_handler_req_lib_cpg_iteration_initialize ( void *conn, const void *message) { const struct req_lib_cpg_iterationinitialize *req_lib_cpg_iterationinitialize = message; struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn); hdb_handle_t cpg_iteration_handle = 0; struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize; struct list_head *iter, *iter2; struct cpg_iteration_instance *cpg_iteration_instance; cs_error_t error = CS_OK; int res; log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration initialize"); /* Because between calling this function and *next can be some operations which will * change list, we must do full copy. */ /* * Create new iteration instance */ res = hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance), &cpg_iteration_handle); if (res != 0) { error = CS_ERR_NO_MEMORY; goto response_send; } res = hdb_handle_get (&cpg_iteration_handle_t_db, cpg_iteration_handle, (void *)&cpg_iteration_instance); if (res != 0) { error = CS_ERR_BAD_HANDLE; goto error_destroy; } list_init (&cpg_iteration_instance->items_list_head); cpg_iteration_instance->handle = cpg_iteration_handle; /* * Create copy of process_info list "grouped by" group name */ for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) { struct process_info *pi = list_entry (iter, struct process_info, list); struct process_info *new_pi; if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) { /* * Try to find processed group name in our list new list */ int found = 0; for (iter2 = cpg_iteration_instance->items_list_head.next; iter2 != &cpg_iteration_instance->items_list_head; iter2 = iter2->next) { struct process_info *pi2 = list_entry (iter2, struct process_info, list); if (mar_name_compare (&pi2->group, &pi->group) == 0) { found = 1; break; } } if (found) { /* * We have this name in list -> don't add */ continue ; } } else if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_ONE_GROUP) { /* * Test pi group name with request */ if (mar_name_compare (&pi->group, &req_lib_cpg_iterationinitialize->group_name) != 0) /* * Not same -> don't add */ continue ; } new_pi = malloc (sizeof (struct process_info)); if (!new_pi) { log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct"); error = CS_ERR_NO_MEMORY; goto error_put_destroy; } memcpy (new_pi, pi, sizeof (struct process_info)); list_init (&new_pi->list); if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) { /* * pid and nodeid -> undefined */ new_pi->pid = new_pi->nodeid = 0; } /* * We will return list "grouped" by "group name", so try to find right place to add */ for (iter2 = cpg_iteration_instance->items_list_head.next; iter2 != &cpg_iteration_instance->items_list_head; iter2 = iter2->next) { struct process_info *pi2 = list_entry (iter2, struct process_info, list); if (mar_name_compare (&pi2->group, &pi->group) == 0) { break; } } list_add (&new_pi->list, iter2); } /* * Now we have a full "grouped by" copy of process_info list */ /* * Add instance to current cpd list */ list_init (&cpg_iteration_instance->list); list_add (&cpg_iteration_instance->list, &cpd->iteration_instance_list_head); cpg_iteration_instance->current_pointer = &cpg_iteration_instance->items_list_head; error_put_destroy: hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_handle); error_destroy: if (error != CS_OK) { hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_handle); } response_send: res_lib_cpg_iterationinitialize.header.size = sizeof (res_lib_cpg_iterationinitialize); res_lib_cpg_iterationinitialize.header.id = MESSAGE_RES_CPG_ITERATIONINITIALIZE; res_lib_cpg_iterationinitialize.header.error = error; res_lib_cpg_iterationinitialize.iteration_handle = cpg_iteration_handle; api->ipc_response_send (conn, &res_lib_cpg_iterationinitialize, sizeof (res_lib_cpg_iterationinitialize)); } static void message_handler_req_lib_cpg_iteration_next ( void *conn, const void *message) { const struct req_lib_cpg_iterationnext *req_lib_cpg_iterationnext = message; struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext; struct cpg_iteration_instance *cpg_iteration_instance; cs_error_t error = CS_OK; int res; struct process_info *pi; log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration next"); res = hdb_handle_get (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle, (void *)&cpg_iteration_instance); if (res != 0) { error = CS_ERR_LIBRARY; goto error_exit; } assert (cpg_iteration_instance); cpg_iteration_instance->current_pointer = cpg_iteration_instance->current_pointer->next; if (cpg_iteration_instance->current_pointer == &cpg_iteration_instance->items_list_head) { error = CS_ERR_NO_SECTIONS; goto error_put; } pi = list_entry (cpg_iteration_instance->current_pointer, struct process_info, list); /* * Copy iteration data */ res_lib_cpg_iterationnext.description.nodeid = pi->nodeid; res_lib_cpg_iterationnext.description.pid = pi->pid; memcpy (&res_lib_cpg_iterationnext.description.group, &pi->group, sizeof (mar_cpg_name_t)); error_put: hdb_handle_put (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle); error_exit: res_lib_cpg_iterationnext.header.size = sizeof (res_lib_cpg_iterationnext); res_lib_cpg_iterationnext.header.id = MESSAGE_RES_CPG_ITERATIONNEXT; res_lib_cpg_iterationnext.header.error = error; api->ipc_response_send (conn, &res_lib_cpg_iterationnext, sizeof (res_lib_cpg_iterationnext)); } static void message_handler_req_lib_cpg_iteration_finalize ( void *conn, const void *message) { const struct req_lib_cpg_iterationfinalize *req_lib_cpg_iterationfinalize = message; struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize; struct cpg_iteration_instance *cpg_iteration_instance; cs_error_t error = CS_OK; int res; log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration finalize"); res = hdb_handle_get (&cpg_iteration_handle_t_db, req_lib_cpg_iterationfinalize->iteration_handle, (void *)&cpg_iteration_instance); if (res != 0) { error = CS_ERR_LIBRARY; goto error_exit; } assert (cpg_iteration_instance); cpg_iteration_instance_finalize (cpg_iteration_instance); hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle); error_exit: res_lib_cpg_iterationfinalize.header.size = sizeof (res_lib_cpg_iterationfinalize); res_lib_cpg_iterationfinalize.header.id = MESSAGE_RES_CPG_ITERATIONFINALIZE; res_lib_cpg_iterationfinalize.header.error = error; api->ipc_response_send (conn, &res_lib_cpg_iterationfinalize, sizeof (res_lib_cpg_iterationfinalize)); } diff --git a/include/corosync/cpg.h b/include/corosync/cpg.h index 55fc4b89..f66fb143 100644 --- a/include/corosync/cpg.h +++ b/include/corosync/cpg.h @@ -1,295 +1,302 @@ /* * Copyright (c) 2006-2011 Red Hat, Inc. * * All rights reserved. * * Author: Christine Caulfield (ccaulfi@redhat.com) * Author: Jan Friesse (jfriesse@redhat.com) * * This software licensed under BSD license, the text of which follows: * * 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. * - Neither the name of the MontaVista Software, Inc. nor the names of its * contributors may 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 OWNER 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. */ #ifndef COROSYNC_CPG_H_DEFINED #define COROSYNC_CPG_H_DEFINED #include #include #ifdef __cplusplus extern "C" { #endif /** * @addtogroup cpg_corosync * * @{ */ typedef uint64_t cpg_handle_t; typedef uint64_t cpg_iteration_handle_t; typedef enum { CPG_TYPE_UNORDERED, /**< not implemented */ CPG_TYPE_FIFO, /**< same as agreed */ CPG_TYPE_AGREED, CPG_TYPE_SAFE /**< not implemented */ } cpg_guarantee_t; typedef enum { CPG_FLOW_CONTROL_DISABLED, /**< flow control is disabled - new messages may be sent */ CPG_FLOW_CONTROL_ENABLED /**< flow control is enabled - new messages should not be sent */ } cpg_flow_control_state_t; typedef enum { CPG_REASON_JOIN = 1, CPG_REASON_LEAVE = 2, CPG_REASON_NODEDOWN = 3, CPG_REASON_NODEUP = 4, CPG_REASON_PROCDOWN = 5 } cpg_reason_t; typedef enum { CPG_ITERATION_NAME_ONLY = 1, CPG_ITERATION_ONE_GROUP = 2, CPG_ITERATION_ALL = 3, } cpg_iteration_type_t; typedef enum { CPG_MODEL_V1 = 1, } cpg_model_t; struct cpg_address { uint32_t nodeid; uint32_t pid; uint32_t reason; }; #define CPG_MAX_NAME_LENGTH 128 struct cpg_name { uint32_t length; char value[CPG_MAX_NAME_LENGTH]; }; #define CPG_MEMBERS_MAX 128 struct cpg_iteration_description_t { struct cpg_name group; uint32_t nodeid; uint32_t pid; }; struct cpg_ring_id { uint32_t nodeid; uint64_t seq; }; typedef void (*cpg_deliver_fn_t) ( cpg_handle_t handle, const struct cpg_name *group_name, uint32_t nodeid, uint32_t pid, /** * Unlike many "msg" pointers, this one is deliberately *not* * declared const in order to permit in-place endian conversion. */ void *msg, size_t msg_len); typedef void (*cpg_confchg_fn_t) ( cpg_handle_t handle, const struct cpg_name *group_name, const struct cpg_address *member_list, size_t member_list_entries, const struct cpg_address *left_list, size_t left_list_entries, const struct cpg_address *joined_list, size_t joined_list_entries); typedef void (*cpg_totem_confchg_fn_t) ( cpg_handle_t handle, struct cpg_ring_id ring_id, uint32_t member_list_entries, const uint32_t *member_list); typedef struct { cpg_deliver_fn_t cpg_deliver_fn; cpg_confchg_fn_t cpg_confchg_fn; } cpg_callbacks_t; typedef struct { cpg_model_t model; } cpg_model_data_t; #define CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF 0x01 typedef struct { cpg_model_t model; cpg_deliver_fn_t cpg_deliver_fn; cpg_confchg_fn_t cpg_confchg_fn; cpg_totem_confchg_fn_t cpg_totem_confchg_fn; unsigned int flags; } cpg_model_v1_data_t; /** @} */ /** * Create a new cpg connection */ cs_error_t cpg_initialize ( cpg_handle_t *handle, cpg_callbacks_t *callbacks); /** * Create a new cpg connection, initialize with model */ cs_error_t cpg_model_initialize ( cpg_handle_t *handle, cpg_model_t model, cpg_model_data_t *model_data, void *context); /** * Close the cpg handle */ cs_error_t cpg_finalize ( cpg_handle_t handle); /** * Get a file descriptor on which to poll. * * cpg_handle_t is NOT a file descriptor and may not be used directly. */ cs_error_t cpg_fd_get ( cpg_handle_t handle, int *fd); +/** + * Get maximum size of a message that will not be fragmented + */ +cs_error_t cpg_max_atomic_msgsize_get ( + cpg_handle_t handle, + uint32_t *size); + /** * Get contexts for a CPG handle */ cs_error_t cpg_context_get ( cpg_handle_t handle, void **context); /** * Set contexts for a CPG handle */ cs_error_t cpg_context_set ( cpg_handle_t handle, void *context); /** * Dispatch messages and configuration changes */ cs_error_t cpg_dispatch ( cpg_handle_t handle, cs_dispatch_flags_t dispatch_types); /** * Join one or more groups. * * messages multicasted with cpg_mcast_joined will be sent to every * group that has been joined on handle handle. Any message multicasted * to a group that has been previously joined will be delivered in cpg_dispatch */ cs_error_t cpg_join ( cpg_handle_t handle, const struct cpg_name *group); /** * Leave one or more groups */ cs_error_t cpg_leave ( cpg_handle_t handle, const struct cpg_name *group); /** * Multicast to groups joined with cpg_join. * * @param handle * @param guarantee * @param iovec This iovec will be multicasted to all groups joined with * the cpg_join interface for handle. * @param iov_len */ cs_error_t cpg_mcast_joined ( cpg_handle_t handle, cpg_guarantee_t guarantee, const struct iovec *iovec, unsigned int iov_len); /** * Get membership information from cpg */ cs_error_t cpg_membership_get ( cpg_handle_t handle, struct cpg_name *groupName, struct cpg_address *member_list, int *member_list_entries); cs_error_t cpg_local_get ( cpg_handle_t handle, unsigned int *local_nodeid); cs_error_t cpg_flow_control_state_get ( cpg_handle_t handle, cpg_flow_control_state_t *flow_control_enabled); cs_error_t cpg_zcb_alloc ( cpg_handle_t handle, size_t size, void **buffer); cs_error_t cpg_zcb_free ( cpg_handle_t handle, void *buffer); cs_error_t cpg_zcb_mcast_joined ( cpg_handle_t handle, cpg_guarantee_t guarantee, void *msg, size_t msg_len); /** * Iteration */ cs_error_t cpg_iteration_initialize( cpg_handle_t handle, cpg_iteration_type_t iteration_type, const struct cpg_name *group, cpg_iteration_handle_t *cpg_iteration_handle); cs_error_t cpg_iteration_next( cpg_iteration_handle_t handle, struct cpg_iteration_description_t *description); cs_error_t cpg_iteration_finalize ( cpg_iteration_handle_t handle); #ifdef __cplusplus } #endif #endif /* COROSYNC_CPG_H_DEFINED */ diff --git a/include/corosync/ipc_cpg.h b/include/corosync/ipc_cpg.h index a95335ad..5008acf5 100644 --- a/include/corosync/ipc_cpg.h +++ b/include/corosync/ipc_cpg.h @@ -1,321 +1,354 @@ /* - * Copyright (c) 2006-2011 Red Hat, Inc. + * Copyright (c) 2006-2015 Red Hat, Inc. * * All rights reserved. * * Author: Christine Caulfield (ccaulfie@redhat.com) * Author: Jan Friesse (jfriesse@redhat.com) * * This software licensed under BSD license, the text of which follows: * * 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. * - Neither the name of the MontaVista Software, Inc. nor the names of its * contributors may 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 OWNER 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. */ #ifndef IPC_CPG_H_DEFINED #define IPC_CPG_H_DEFINED #include #include #include #include #define CPG_ZC_PATH_LEN 128 enum req_cpg_types { MESSAGE_REQ_CPG_JOIN = 0, MESSAGE_REQ_CPG_LEAVE = 1, MESSAGE_REQ_CPG_MCAST = 2, MESSAGE_REQ_CPG_MEMBERSHIP = 3, MESSAGE_REQ_CPG_LOCAL_GET = 4, MESSAGE_REQ_CPG_ITERATIONINITIALIZE = 5, MESSAGE_REQ_CPG_ITERATIONNEXT = 6, MESSAGE_REQ_CPG_ITERATIONFINALIZE = 7, MESSAGE_REQ_CPG_FINALIZE = 8, MESSAGE_REQ_CPG_ZC_ALLOC = 9, MESSAGE_REQ_CPG_ZC_FREE = 10, MESSAGE_REQ_CPG_ZC_EXECUTE = 11, + MESSAGE_REQ_CPG_PARTIAL_MCAST = 12, }; enum res_cpg_types { MESSAGE_RES_CPG_JOIN = 0, MESSAGE_RES_CPG_LEAVE = 1, MESSAGE_RES_CPG_MCAST = 2, MESSAGE_RES_CPG_MEMBERSHIP = 3, MESSAGE_RES_CPG_CONFCHG_CALLBACK = 4, MESSAGE_RES_CPG_DELIVER_CALLBACK = 5, MESSAGE_RES_CPG_FLOW_CONTROL_STATE_SET = 6, MESSAGE_RES_CPG_LOCAL_GET = 7, MESSAGE_RES_CPG_FLOWCONTROL_CALLBACK = 8, MESSAGE_RES_CPG_ITERATIONINITIALIZE = 9, MESSAGE_RES_CPG_ITERATIONNEXT = 10, MESSAGE_RES_CPG_ITERATIONFINALIZE = 11, MESSAGE_RES_CPG_FINALIZE = 12, MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK = 13, MESSAGE_RES_CPG_ZC_ALLOC = 14, MESSAGE_RES_CPG_ZC_FREE = 15, MESSAGE_RES_CPG_ZC_EXECUTE = 16, + MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK = 17, + MESSAGE_RES_CPG_PARTIAL_SEND = 18, }; enum lib_cpg_confchg_reason { CONFCHG_CPG_REASON_JOIN = 1, CONFCHG_CPG_REASON_LEAVE = 2, CONFCHG_CPG_REASON_NODEDOWN = 3, CONFCHG_CPG_REASON_NODEUP = 4, CONFCHG_CPG_REASON_PROCDOWN = 5 }; +enum lib_cpg_partial_types { + LIBCPG_PARTIAL_FIRST = 1, + LIBCPG_PARTIAL_CONTINUED = 2, + LIBCPG_PARTIAL_LAST = 3, +}; + typedef struct { uint32_t length __attribute__((aligned(8))); char value[CPG_MAX_NAME_LENGTH] __attribute__((aligned(8))); } mar_cpg_name_t; static inline void swab_mar_cpg_name_t (mar_cpg_name_t *to_swab) { swab_mar_uint32_t (&to_swab->length); } static inline void marshall_from_mar_cpg_name_t ( struct cpg_name *dest, const mar_cpg_name_t *src) { dest->length = src->length; memcpy (&dest->value, &src->value, CPG_MAX_NAME_LENGTH); } static inline void marshall_to_mar_cpg_name_t ( mar_cpg_name_t *dest, const struct cpg_name *src) { dest->length = src->length; memcpy (&dest->value, &src->value, CPG_MAX_NAME_LENGTH); } typedef struct { mar_uint32_t nodeid __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); mar_uint32_t reason __attribute__((aligned(8))); } mar_cpg_address_t; static inline void marshall_from_mar_cpg_address_t ( struct cpg_address *dest, const mar_cpg_address_t *src) { dest->nodeid = src->nodeid; dest->pid = src->pid; dest->reason = src->reason; } static inline void marshall_to_mar_cpg_address_t ( mar_cpg_address_t *dest, const struct cpg_address *src) { dest->nodeid = src->nodeid; dest->pid = src->pid; dest->reason = src->reason; } static inline int mar_name_compare ( const mar_cpg_name_t *g1, const mar_cpg_name_t *g2) { return (g1->length == g2->length? memcmp (g1->value, g2->value, g1->length): g1->length - g2->length); } typedef struct { mar_cpg_name_t group; mar_uint32_t nodeid; mar_uint32_t pid; } mar_cpg_iteration_description_t; static inline void marshall_from_mar_cpg_iteration_description_t( struct cpg_iteration_description_t *dest, const mar_cpg_iteration_description_t *src) { dest->nodeid = src->nodeid; dest->pid = src->pid; marshall_from_mar_cpg_name_t (&dest->group, &src->group); }; typedef struct { mar_uint32_t nodeid __attribute__((aligned(8))); mar_uint64_t seq __attribute__((aligned(8))); } mar_cpg_ring_id_t; static inline void marshall_from_mar_cpg_ring_id_t ( struct cpg_ring_id *dest, const mar_cpg_ring_id_t *src) { dest->nodeid = src->nodeid; dest->seq = src->seq; } struct req_lib_cpg_join { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); mar_uint32_t flags __attribute__((aligned(8))); }; struct res_lib_cpg_join { struct qb_ipc_response_header header __attribute__((aligned(8))); }; struct req_lib_cpg_finalize { struct qb_ipc_request_header header __attribute__((aligned(8))); }; struct res_lib_cpg_finalize { struct qb_ipc_response_header header __attribute__((aligned(8))); }; struct req_lib_cpg_local_get { struct qb_ipc_request_header header __attribute__((aligned(8))); }; struct res_lib_cpg_local_get { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_uint32_t local_nodeid __attribute__((aligned(8))); }; +struct res_lib_cpg_partial_send { + struct qb_ipc_response_header header __attribute__((aligned(8))); +}; + struct req_lib_cpg_mcast { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_uint32_t guarantee __attribute__((aligned(8))); mar_uint32_t msglen __attribute__((aligned(8))); mar_uint8_t message[] __attribute__((aligned(8))); }; +struct req_lib_cpg_partial_mcast { + struct qb_ipc_response_header header __attribute__((aligned(8))); + mar_uint32_t guarantee __attribute__((aligned(8))); + mar_uint32_t msglen __attribute__((aligned(8))); + mar_uint32_t fraglen __attribute__((aligned(8))); + mar_uint32_t type __attribute__((aligned(8))); + mar_uint8_t message[] __attribute__((aligned(8))); +}; + struct res_lib_cpg_mcast { struct qb_ipc_response_header header __attribute__((aligned(8))); }; /** * Message from another node */ struct res_lib_cpg_deliver_callback { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t msglen __attribute__((aligned(8))); mar_uint32_t nodeid __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); mar_uint8_t message[] __attribute__((aligned(8))); }; +struct res_lib_cpg_partial_deliver_callback { + struct qb_ipc_response_header header __attribute__((aligned(8))); + mar_cpg_name_t group_name __attribute__((aligned(8))); + mar_uint32_t msglen __attribute__((aligned(8))); + mar_uint32_t fraglen __attribute__((aligned(8))); + mar_uint32_t nodeid __attribute__((aligned(8))); + mar_uint32_t pid __attribute__((aligned(8))); + mar_uint32_t type __attribute__((aligned(8))); + mar_uint8_t message[] __attribute__((aligned(8))); +}; + struct res_lib_cpg_flowcontrol_callback { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_uint32_t flow_control_state __attribute__((aligned(8))); }; struct req_lib_cpg_membership_get { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); }; struct res_lib_cpg_membership_get { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_uint32_t member_count __attribute__((aligned(8))); mar_cpg_address_t member_list[PROCESSOR_COUNT_MAX]; }; struct res_lib_cpg_confchg_callback { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t member_list_entries __attribute__((aligned(8))); mar_uint32_t joined_list_entries __attribute__((aligned(8))); mar_uint32_t left_list_entries __attribute__((aligned(8))); mar_cpg_address_t member_list[]; // struct cpg_address left_list[]; // struct cpg_address joined_list[]; }; struct res_lib_cpg_totem_confchg_callback { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_cpg_ring_id_t ring_id __attribute__((aligned(8))); mar_uint32_t member_list_entries __attribute__((aligned(8))); mar_uint32_t member_list[]; }; struct req_lib_cpg_leave { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t pid __attribute__((aligned(8))); }; struct res_lib_cpg_leave { struct qb_ipc_response_header header __attribute__((aligned(8))); }; struct req_lib_cpg_iterationinitialize { struct qb_ipc_request_header header __attribute__((aligned(8))); mar_cpg_name_t group_name __attribute__((aligned(8))); mar_uint32_t iteration_type __attribute__((aligned(8))); }; struct res_lib_cpg_iterationinitialize { struct qb_ipc_response_header header __attribute__((aligned(8))); hdb_handle_t iteration_handle __attribute__((aligned(8))); }; struct req_lib_cpg_iterationnext { struct qb_ipc_request_header header __attribute__((aligned(8))); hdb_handle_t iteration_handle __attribute__((aligned(8))); }; struct res_lib_cpg_iterationnext { struct qb_ipc_response_header header __attribute__((aligned(8))); mar_cpg_iteration_description_t description __attribute__((aligned(8))); }; struct req_lib_cpg_iterationfinalize { struct qb_ipc_request_header header __attribute__((aligned(8))); hdb_handle_t iteration_handle __attribute__((aligned(8))); }; struct res_lib_cpg_iterationfinalize { struct qb_ipc_response_header header __attribute__((aligned(8))); }; typedef struct { struct qb_ipc_request_header header __attribute__((aligned(8))); size_t map_size __attribute__((aligned(8))); char path_to_file[CPG_ZC_PATH_LEN] __attribute__((aligned(8))); } mar_req_coroipcc_zc_alloc_t __attribute__((aligned(8))); typedef struct { struct qb_ipc_request_header header __attribute__((aligned(8))); size_t map_size __attribute__((aligned(8))); uint64_t server_address __attribute__((aligned(8))); } mar_req_coroipcc_zc_free_t __attribute__((aligned(8))); typedef struct { struct qb_ipc_request_header header __attribute__((aligned(8))); uint64_t server_address __attribute__((aligned(8))); } mar_req_coroipcc_zc_execute_t __attribute__((aligned(8))); struct coroipcs_zc_header { int map_size; uint64_t server_address; }; #endif /* IPC_CPG_H_DEFINED */ diff --git a/lib/cpg.c b/lib/cpg.c index 4b92f44b..037e8a93 100644 --- a/lib/cpg.c +++ b/lib/cpg.c @@ -1,1188 +1,1355 @@ /* * vi: set autoindent tabstop=4 shiftwidth=4 : * - * Copyright (c) 2006-2012 Red Hat, Inc. + * Copyright (c) 2006-2015 Red Hat, Inc. * * All rights reserved. * * Author: Christine Caulfield (ccaulfi@redhat.com) * Author: Jan Friesse (jfriesse@redhat.com) * * This software licensed under BSD license, the text of which follows: * * 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. * - Neither the name of the MontaVista Software, Inc. nor the names of its * contributors may 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 OWNER 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. */ /* * Provides a closed process group API using the coroipcc executive */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "util.h" #ifndef MAP_ANONYMOUS #define MAP_ANONYMOUS MAP_ANON #endif +/* + * Maximum number of times to retry a send when transmitting + * a large message fragment + */ +#define MAX_RETRIES 100 + /* * ZCB files have following umask (umask is same as used in libqb) */ #define CPG_MEMORY_MAP_UMASK 077 struct cpg_inst { qb_ipcc_connection_t *c; int finalize; void *context; union { cpg_model_data_t model_data; cpg_model_v1_data_t model_v1_data; }; struct list_head iteration_list_head; + uint32_t max_msg_size; + char *assembly_buf; + uint32_t assembly_buf_ptr; + int assembling; /* Flag that says we have started assembling a message. + * It's here to catch the situation where a node joins + * the cluster/group in the middle of a CPG message send + * so we don't pass on a partial message to the client. + */ }; static void cpg_inst_free (void *inst); DECLARE_HDB_DATABASE(cpg_handle_t_db, cpg_inst_free); struct cpg_iteration_instance_t { cpg_iteration_handle_t cpg_iteration_handle; qb_ipcc_connection_t *conn; hdb_handle_t executive_iteration_handle; struct list_head list; }; DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL); /* * Internal (not visible by API) functions */ static cs_error_t coroipcc_msg_send_reply_receive ( qb_ipcc_connection_t *c, const struct iovec *iov, unsigned int iov_len, void *res_msg, size_t res_len) { return qb_to_cs_error(qb_ipcc_sendv_recv(c, iov, iov_len, res_msg, res_len, CS_IPC_TIMEOUT_MS)); } static void cpg_iteration_instance_finalize (struct cpg_iteration_instance_t *cpg_iteration_instance) { list_del (&cpg_iteration_instance->list); hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle); } static void cpg_inst_free (void *inst) { struct cpg_inst *cpg_inst = (struct cpg_inst *)inst; qb_ipcc_disconnect(cpg_inst->c); } static void cpg_inst_finalize (struct cpg_inst *cpg_inst, hdb_handle_t handle) { struct list_head *iter, *iter_next; struct cpg_iteration_instance_t *cpg_iteration_instance; /* * Traverse thru iteration instances and delete them */ for (iter = cpg_inst->iteration_list_head.next; iter != &cpg_inst->iteration_list_head;iter = iter_next) { iter_next = iter->next; cpg_iteration_instance = list_entry (iter, struct cpg_iteration_instance_t, list); cpg_iteration_instance_finalize (cpg_iteration_instance); } hdb_handle_destroy (&cpg_handle_t_db, handle); } /** * @defgroup cpg_coroipcc The closed process group API * @ingroup coroipcc * * @{ */ cs_error_t cpg_initialize ( cpg_handle_t *handle, cpg_callbacks_t *callbacks) { cpg_model_v1_data_t model_v1_data; memset (&model_v1_data, 0, sizeof (cpg_model_v1_data_t)); if (callbacks) { model_v1_data.cpg_deliver_fn = callbacks->cpg_deliver_fn; model_v1_data.cpg_confchg_fn = callbacks->cpg_confchg_fn; } return (cpg_model_initialize (handle, CPG_MODEL_V1, (cpg_model_data_t *)&model_v1_data, NULL)); } cs_error_t cpg_model_initialize ( cpg_handle_t *handle, cpg_model_t model, cpg_model_data_t *model_data, void *context) { cs_error_t error; struct cpg_inst *cpg_inst; if (model != CPG_MODEL_V1) { error = CS_ERR_INVALID_PARAM; goto error_no_destroy; } error = hdb_error_to_cs (hdb_handle_create (&cpg_handle_t_db, sizeof (struct cpg_inst), handle)); if (error != CS_OK) { goto error_no_destroy; } error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, *handle, (void *)&cpg_inst)); if (error != CS_OK) { goto error_destroy; } cpg_inst->c = qb_ipcc_connect ("cpg", IPC_REQUEST_SIZE); if (cpg_inst->c == NULL) { error = qb_to_cs_error(-errno); goto error_put_destroy; } if (model_data != NULL) { switch (model) { case CPG_MODEL_V1: memcpy (&cpg_inst->model_v1_data, model_data, sizeof (cpg_model_v1_data_t)); if ((cpg_inst->model_v1_data.flags & ~(CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF)) != 0) { error = CS_ERR_INVALID_PARAM; goto error_destroy; } break; } } + /* Allow space for corosync internal headers */ + cpg_inst->max_msg_size = IPC_REQUEST_SIZE - 1024; cpg_inst->model_data.model = model; cpg_inst->context = context; list_init(&cpg_inst->iteration_list_head); hdb_handle_put (&cpg_handle_t_db, *handle); return (CS_OK); error_put_destroy: hdb_handle_put (&cpg_handle_t_db, *handle); error_destroy: hdb_handle_destroy (&cpg_handle_t_db, *handle); error_no_destroy: return (error); } cs_error_t cpg_finalize ( cpg_handle_t handle) { struct cpg_inst *cpg_inst; struct iovec iov; struct req_lib_cpg_finalize req_lib_cpg_finalize; struct res_lib_cpg_finalize res_lib_cpg_finalize; cs_error_t error; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } /* * Another thread has already started finalizing */ if (cpg_inst->finalize) { hdb_handle_put (&cpg_handle_t_db, handle); return (CS_ERR_BAD_HANDLE); } cpg_inst->finalize = 1; /* * Send service request */ req_lib_cpg_finalize.header.size = sizeof (struct req_lib_cpg_finalize); req_lib_cpg_finalize.header.id = MESSAGE_REQ_CPG_FINALIZE; iov.iov_base = (void *)&req_lib_cpg_finalize; iov.iov_len = sizeof (struct req_lib_cpg_finalize); error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1, &res_lib_cpg_finalize, sizeof (struct res_lib_cpg_finalize)); cpg_inst_finalize (cpg_inst, handle); hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_fd_get ( cpg_handle_t handle, int *fd) { cs_error_t error; struct cpg_inst *cpg_inst; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } error = qb_to_cs_error (qb_ipcc_fd_get (cpg_inst->c, fd)); hdb_handle_put (&cpg_handle_t_db, handle); return (error); } +cs_error_t cpg_max_atomic_msgsize_get ( + cpg_handle_t handle, + uint32_t *size) +{ + cs_error_t error; + struct cpg_inst *cpg_inst; + + error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); + if (error != CS_OK) { + return (error); + } + + *size = cpg_inst->max_msg_size; + + hdb_handle_put (&cpg_handle_t_db, handle); + + return (error); +} + cs_error_t cpg_context_get ( cpg_handle_t handle, void **context) { cs_error_t error; struct cpg_inst *cpg_inst; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } *context = cpg_inst->context; hdb_handle_put (&cpg_handle_t_db, handle); return (CS_OK); } cs_error_t cpg_context_set ( cpg_handle_t handle, void *context) { cs_error_t error; struct cpg_inst *cpg_inst; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } cpg_inst->context = context; hdb_handle_put (&cpg_handle_t_db, handle); return (CS_OK); } cs_error_t cpg_dispatch ( cpg_handle_t handle, cs_dispatch_flags_t dispatch_types) { int timeout = -1; cs_error_t error; int cont = 1; /* always continue do loop except when set to 0 */ struct cpg_inst *cpg_inst; struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback; struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback; + struct res_lib_cpg_partial_deliver_callback *res_cpg_partial_deliver_callback; struct res_lib_cpg_totem_confchg_callback *res_cpg_totem_confchg_callback; struct cpg_inst cpg_inst_copy; struct qb_ipc_response_header *dispatch_data; struct cpg_address member_list[CPG_MEMBERS_MAX]; struct cpg_address left_list[CPG_MEMBERS_MAX]; struct cpg_address joined_list[CPG_MEMBERS_MAX]; struct cpg_name group_name; mar_cpg_address_t *left_list_start; mar_cpg_address_t *joined_list_start; unsigned int i; struct cpg_ring_id ring_id; uint32_t totem_member_list[CPG_MEMBERS_MAX]; int32_t errno_res; char dispatch_buf[IPC_DISPATCH_SIZE]; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } /* * Timeout instantly for CS_DISPATCH_ONE_NONBLOCKING or CS_DISPATCH_ALL and - * wait indefinately for CS_DISPATCH_ONE or CS_DISPATCH_BLOCKING + * wait indefinitely for CS_DISPATCH_ONE or CS_DISPATCH_BLOCKING */ if (dispatch_types == CS_DISPATCH_ALL || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) { timeout = 0; } dispatch_data = (struct qb_ipc_response_header *)dispatch_buf; do { errno_res = qb_ipcc_event_recv ( cpg_inst->c, dispatch_buf, IPC_DISPATCH_SIZE, timeout); error = qb_to_cs_error (errno_res); if (error == CS_ERR_BAD_HANDLE) { error = CS_OK; goto error_put; } if (error == CS_ERR_TRY_AGAIN) { if (dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) { /* * Don't mask error */ goto error_put; } error = CS_OK; if (dispatch_types == CS_DISPATCH_ALL) { break; /* exit do while cont is 1 loop */ } else { continue; /* next poll */ } } if (error != CS_OK) { goto error_put; } /* * Make copy of callbacks, message data, unlock instance, and call callback * A risk of this dispatch method is that the callback routines may * operate at the same time that cpgFinalize has been called. */ memcpy (&cpg_inst_copy, cpg_inst, sizeof (struct cpg_inst)); switch (cpg_inst_copy.model_data.model) { case CPG_MODEL_V1: /* * Dispatch incoming message */ switch (dispatch_data->id) { case MESSAGE_RES_CPG_DELIVER_CALLBACK: if (cpg_inst_copy.model_v1_data.cpg_deliver_fn == NULL) { break; } res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data; marshall_from_mar_cpg_name_t ( &group_name, &res_cpg_deliver_callback->group_name); cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle, &group_name, res_cpg_deliver_callback->nodeid, res_cpg_deliver_callback->pid, &res_cpg_deliver_callback->message, res_cpg_deliver_callback->msglen); break; + case MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK: + res_cpg_partial_deliver_callback = (struct res_lib_cpg_partial_deliver_callback *)dispatch_data; + + marshall_from_mar_cpg_name_t ( + &group_name, + &res_cpg_partial_deliver_callback->group_name); + + if (res_cpg_partial_deliver_callback->type == LIBCPG_PARTIAL_FIRST) { + /* + * Allocate a buffer to contain a full message. + */ + cpg_inst->assembly_buf = malloc(res_cpg_partial_deliver_callback->msglen); + if (!cpg_inst->assembly_buf) { + error = CS_ERR_NO_MEMORY; + goto error_put; + } + cpg_inst->assembling = 1; + cpg_inst->assembly_buf_ptr = 0; + } + if (cpg_inst->assembling) { + memcpy(cpg_inst->assembly_buf + cpg_inst->assembly_buf_ptr, + res_cpg_partial_deliver_callback->message, res_cpg_partial_deliver_callback->fraglen); + cpg_inst->assembly_buf_ptr += res_cpg_partial_deliver_callback->fraglen; + + if (res_cpg_partial_deliver_callback->type == LIBCPG_PARTIAL_LAST) { + cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle, + &group_name, + res_cpg_partial_deliver_callback->nodeid, + res_cpg_partial_deliver_callback->pid, + cpg_inst->assembly_buf, + res_cpg_partial_deliver_callback->msglen); + free(cpg_inst->assembly_buf); + cpg_inst->assembling = 0; + } + } + break; + case MESSAGE_RES_CPG_CONFCHG_CALLBACK: if (cpg_inst_copy.model_v1_data.cpg_confchg_fn == NULL) { break; } res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data; for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) { marshall_from_mar_cpg_address_t (&member_list[i], &res_cpg_confchg_callback->member_list[i]); } left_list_start = res_cpg_confchg_callback->member_list + res_cpg_confchg_callback->member_list_entries; for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) { marshall_from_mar_cpg_address_t (&left_list[i], &left_list_start[i]); } joined_list_start = res_cpg_confchg_callback->member_list + res_cpg_confchg_callback->member_list_entries + res_cpg_confchg_callback->left_list_entries; for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) { marshall_from_mar_cpg_address_t (&joined_list[i], &joined_list_start[i]); } marshall_from_mar_cpg_name_t ( &group_name, &res_cpg_confchg_callback->group_name); cpg_inst_copy.model_v1_data.cpg_confchg_fn (handle, &group_name, member_list, res_cpg_confchg_callback->member_list_entries, left_list, res_cpg_confchg_callback->left_list_entries, joined_list, res_cpg_confchg_callback->joined_list_entries); break; case MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK: if (cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn == NULL) { break; } res_cpg_totem_confchg_callback = (struct res_lib_cpg_totem_confchg_callback *)dispatch_data; marshall_from_mar_cpg_ring_id_t (&ring_id, &res_cpg_totem_confchg_callback->ring_id); for (i = 0; i < res_cpg_totem_confchg_callback->member_list_entries; i++) { totem_member_list[i] = res_cpg_totem_confchg_callback->member_list[i]; } cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn (handle, ring_id, res_cpg_totem_confchg_callback->member_list_entries, totem_member_list); break; default: error = CS_ERR_LIBRARY; goto error_put; break; } /* - switch (dispatch_data->id) */ break; /* case CPG_MODEL_V1 */ } /* - switch (cpg_inst_copy.model_data.model) */ if (cpg_inst_copy.finalize || cpg_inst->finalize) { /* * If the finalize has been called then get out of the dispatch. */ cpg_inst->finalize = 1; error = CS_ERR_BAD_HANDLE; goto error_put; } /* * Determine if more messages should be processed */ if (dispatch_types == CS_DISPATCH_ONE || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) { cont = 0; } } while (cont); error_put: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_join ( cpg_handle_t handle, const struct cpg_name *group) { cs_error_t error; struct cpg_inst *cpg_inst; struct iovec iov[2]; struct req_lib_cpg_join req_lib_cpg_join; struct res_lib_cpg_join response; if (group->length > CPG_MAX_NAME_LENGTH) { return (CS_ERR_NAME_TOO_LONG); } error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } /* Now join */ req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join); req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN; req_lib_cpg_join.pid = getpid(); req_lib_cpg_join.flags = 0; switch (cpg_inst->model_data.model) { case CPG_MODEL_V1: req_lib_cpg_join.flags = cpg_inst->model_v1_data.flags; break; } marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name, group); iov[0].iov_base = (void *)&req_lib_cpg_join; iov[0].iov_len = sizeof (struct req_lib_cpg_join); do { error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1, &response, sizeof (struct res_lib_cpg_join)); if (error != CS_OK) { goto error_exit; } } while (response.header.error == CS_ERR_BUSY); error = response.header.error; error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_leave ( cpg_handle_t handle, const struct cpg_name *group) { cs_error_t error; struct cpg_inst *cpg_inst; struct iovec iov[2]; struct req_lib_cpg_leave req_lib_cpg_leave; struct res_lib_cpg_leave res_lib_cpg_leave; if (group->length > CPG_MAX_NAME_LENGTH) { return (CS_ERR_NAME_TOO_LONG); } error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave); req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE; req_lib_cpg_leave.pid = getpid(); marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name, group); iov[0].iov_base = (void *)&req_lib_cpg_leave; iov[0].iov_len = sizeof (struct req_lib_cpg_leave); do { error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1, &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave)); if (error != CS_OK) { goto error_exit; } } while (res_lib_cpg_leave.header.error == CS_ERR_BUSY); error = res_lib_cpg_leave.header.error; error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_membership_get ( cpg_handle_t handle, struct cpg_name *group_name, struct cpg_address *member_list, int *member_list_entries) { cs_error_t error; struct cpg_inst *cpg_inst; struct iovec iov; struct req_lib_cpg_membership_get req_lib_cpg_membership_get; struct res_lib_cpg_membership_get res_lib_cpg_membership_get; unsigned int i; if (group_name->length > CPG_MAX_NAME_LENGTH) { return (CS_ERR_NAME_TOO_LONG); } if (member_list == NULL) { return (CS_ERR_INVALID_PARAM); } if (member_list_entries == NULL) { return (CS_ERR_INVALID_PARAM); } error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } req_lib_cpg_membership_get.header.size = sizeof (struct req_lib_cpg_membership_get); req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP; marshall_to_mar_cpg_name_t (&req_lib_cpg_membership_get.group_name, group_name); iov.iov_base = (void *)&req_lib_cpg_membership_get; iov.iov_len = sizeof (struct req_lib_cpg_membership_get); error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1, &res_lib_cpg_membership_get, sizeof (res_lib_cpg_membership_get)); if (error != CS_OK) { goto error_exit; } error = res_lib_cpg_membership_get.header.error; /* * Copy results to caller */ *member_list_entries = res_lib_cpg_membership_get.member_count; if (member_list) { for (i = 0; i < res_lib_cpg_membership_get.member_count; i++) { marshall_from_mar_cpg_address_t (&member_list[i], &res_lib_cpg_membership_get.member_list[i]); } } error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_local_get ( cpg_handle_t handle, unsigned int *local_nodeid) { cs_error_t error; struct cpg_inst *cpg_inst; struct iovec iov; struct req_lib_cpg_local_get req_lib_cpg_local_get; struct res_lib_cpg_local_get res_lib_cpg_local_get; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } req_lib_cpg_local_get.header.size = sizeof (struct qb_ipc_request_header); req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET; iov.iov_base = (void *)&req_lib_cpg_local_get; iov.iov_len = sizeof (struct req_lib_cpg_local_get); error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1, &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get)); if (error != CS_OK) { goto error_exit; } error = res_lib_cpg_local_get.header.error; *local_nodeid = res_lib_cpg_local_get.local_nodeid; error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_flow_control_state_get ( cpg_handle_t handle, cpg_flow_control_state_t *flow_control_state) { cs_error_t error; struct cpg_inst *cpg_inst; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } *flow_control_state = CPG_FLOW_CONTROL_DISABLED; error = CS_OK; hdb_handle_put (&cpg_handle_t_db, handle); return (error); } static int memory_map (char *path, const char *file, void **buf, size_t bytes) { int32_t fd; void *addr; int32_t res; char *buffer; int32_t i; size_t written; size_t page_size; long int sysconf_page_size; mode_t old_umask; snprintf (path, PATH_MAX, "/dev/shm/%s", file); old_umask = umask(CPG_MEMORY_MAP_UMASK); fd = mkstemp (path); (void)umask(old_umask); if (fd == -1) { snprintf (path, PATH_MAX, LOCALSTATEDIR "/run/%s", file); old_umask = umask(CPG_MEMORY_MAP_UMASK); fd = mkstemp (path); (void)umask(old_umask); if (fd == -1) { return (-1); } } res = ftruncate (fd, bytes); if (res == -1) { goto error_close_unlink; } sysconf_page_size = sysconf(_SC_PAGESIZE); if (sysconf_page_size <= 0) { goto error_close_unlink; } page_size = sysconf_page_size; buffer = malloc (page_size); if (buffer == NULL) { goto error_close_unlink; } memset (buffer, 0, page_size); for (i = 0; i < (bytes / page_size); i++) { retry_write: written = write (fd, buffer, page_size); if (written == -1 && errno == EINTR) { goto retry_write; } if (written != page_size) { free (buffer); goto error_close_unlink; } } free (buffer); addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); if (addr == MAP_FAILED) { goto error_close_unlink; } #ifdef MADV_NOSYNC madvise(addr, bytes, MADV_NOSYNC); #endif res = close (fd); if (res) { return (-1); } *buf = addr; return 0; error_close_unlink: close (fd); unlink(path); return -1; } cs_error_t cpg_zcb_alloc ( cpg_handle_t handle, size_t size, void **buffer) { void *buf = NULL; char path[PATH_MAX]; mar_req_coroipcc_zc_alloc_t req_coroipcc_zc_alloc; struct qb_ipc_response_header res_coroipcs_zc_alloc; size_t map_size; struct iovec iovec; struct coroipcs_zc_header *hdr; cs_error_t error; struct cpg_inst *cpg_inst; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } map_size = size + sizeof (struct req_lib_cpg_mcast) + sizeof (struct coroipcs_zc_header); assert(memory_map (path, "corosync_zerocopy-XXXXXX", &buf, map_size) != -1); if (strlen(path) >= CPG_ZC_PATH_LEN) { unlink(path); munmap (buf, map_size); return (CS_ERR_NAME_TOO_LONG); } req_coroipcc_zc_alloc.header.size = sizeof (mar_req_coroipcc_zc_alloc_t); req_coroipcc_zc_alloc.header.id = MESSAGE_REQ_CPG_ZC_ALLOC; req_coroipcc_zc_alloc.map_size = map_size; strcpy (req_coroipcc_zc_alloc.path_to_file, path); iovec.iov_base = (void *)&req_coroipcc_zc_alloc; iovec.iov_len = sizeof (mar_req_coroipcc_zc_alloc_t); error = coroipcc_msg_send_reply_receive ( cpg_inst->c, &iovec, 1, &res_coroipcs_zc_alloc, sizeof (struct qb_ipc_response_header)); hdr = (struct coroipcs_zc_header *)buf; hdr->map_size = map_size; *buffer = ((char *)buf) + sizeof (struct coroipcs_zc_header); hdb_handle_put (&cpg_handle_t_db, handle); *buffer = ((char *)*buffer) + sizeof (struct req_lib_cpg_mcast); return (error); } cs_error_t cpg_zcb_free ( cpg_handle_t handle, void *buffer) { cs_error_t error; struct cpg_inst *cpg_inst; mar_req_coroipcc_zc_free_t req_coroipcc_zc_free; struct qb_ipc_response_header res_coroipcs_zc_free; struct iovec iovec; struct coroipcs_zc_header *header = (struct coroipcs_zc_header *)((char *)buffer - sizeof (struct coroipcs_zc_header)); error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } req_coroipcc_zc_free.header.size = sizeof (mar_req_coroipcc_zc_free_t); req_coroipcc_zc_free.header.id = MESSAGE_REQ_CPG_ZC_FREE; req_coroipcc_zc_free.map_size = header->map_size; req_coroipcc_zc_free.server_address = header->server_address; iovec.iov_base = (void *)&req_coroipcc_zc_free; iovec.iov_len = sizeof (mar_req_coroipcc_zc_free_t); error = coroipcc_msg_send_reply_receive ( cpg_inst->c, &iovec, 1, &res_coroipcs_zc_free, sizeof (struct qb_ipc_response_header)); munmap ((void *)header, header->map_size); hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_zcb_mcast_joined ( cpg_handle_t handle, cpg_guarantee_t guarantee, void *msg, size_t msg_len) { cs_error_t error; struct cpg_inst *cpg_inst; struct req_lib_cpg_mcast *req_lib_cpg_mcast; struct res_lib_cpg_mcast res_lib_cpg_mcast; mar_req_coroipcc_zc_execute_t req_coroipcc_zc_execute; struct coroipcs_zc_header *hdr; struct iovec iovec; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } + + if (msg_len > IPC_REQUEST_SIZE) { + error = CS_ERR_TOO_BIG; + goto error_exit; + } + req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast)); req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) + msg_len; req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST; req_lib_cpg_mcast->guarantee = guarantee; req_lib_cpg_mcast->msglen = msg_len; hdr = (struct coroipcs_zc_header *)(((char *)req_lib_cpg_mcast) - sizeof (struct coroipcs_zc_header)); req_coroipcc_zc_execute.header.size = sizeof (mar_req_coroipcc_zc_execute_t); req_coroipcc_zc_execute.header.id = MESSAGE_REQ_CPG_ZC_EXECUTE; req_coroipcc_zc_execute.server_address = hdr->server_address; iovec.iov_base = (void *)&req_coroipcc_zc_execute; iovec.iov_len = sizeof (mar_req_coroipcc_zc_execute_t); error = coroipcc_msg_send_reply_receive ( cpg_inst->c, &iovec, 1, &res_lib_cpg_mcast, sizeof(res_lib_cpg_mcast)); if (error != CS_OK) { goto error_exit; } error = res_lib_cpg_mcast.header.error; error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } +static cs_error_t send_fragments ( + struct cpg_inst *cpg_inst, + cpg_guarantee_t guarantee, + size_t msg_len, + const struct iovec *iovec, + unsigned int iov_len) +{ + int i; + cs_error_t error = CS_OK; + struct iovec iov[2]; + struct req_lib_cpg_partial_mcast req_lib_cpg_mcast; + struct res_lib_cpg_partial_send res_lib_cpg_partial_send; + size_t sent = 0; + size_t iov_sent = 0; + int retry_count; + + req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_PARTIAL_MCAST; + req_lib_cpg_mcast.guarantee = guarantee; + req_lib_cpg_mcast.msglen = msg_len; + + iov[0].iov_base = (void *)&req_lib_cpg_mcast; + iov[0].iov_len = sizeof (struct req_lib_cpg_partial_mcast); + + i=0; + iov_sent = 0 ; + qb_ipcc_fc_enable_max_set(cpg_inst->c, 2); + + while (error == CS_OK && sent < msg_len) { + + retry_count = 0; + if ( (iovec[i].iov_len - iov_sent) > cpg_inst->max_msg_size) { + iov[1].iov_len = cpg_inst->max_msg_size; + } + else { + iov[1].iov_len = iovec[i].iov_len - iov_sent; + } + + if (sent == 0) { + req_lib_cpg_mcast.type = LIBCPG_PARTIAL_FIRST; + } + else if ((sent + iov[1].iov_len) == msg_len) { + req_lib_cpg_mcast.type = LIBCPG_PARTIAL_LAST; + } + else { + req_lib_cpg_mcast.type = LIBCPG_PARTIAL_CONTINUED; + } + + req_lib_cpg_mcast.fraglen = iov[1].iov_len; + req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_partial_mcast) + iov[1].iov_len; + iov[1].iov_base = (char *)iovec[i].iov_base + iov_sent; + + resend: + error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 2, + &res_lib_cpg_partial_send, + sizeof (res_lib_cpg_partial_send)); + + if (error == CS_ERR_TRY_AGAIN) { + fprintf(stderr, "sleep. counter=%d\n", retry_count); + if (++retry_count > MAX_RETRIES) { + goto error_exit; + } + usleep(10000); + goto resend; + } + + iov_sent += iov[1].iov_len; + sent += iov[1].iov_len; + + /* Next iovec */ + if (iov_sent >= iovec[i].iov_len) { + i++; + iov_sent = 0; + } + error = res_lib_cpg_partial_send.header.error; + } +error_exit: + qb_ipcc_fc_enable_max_set(cpg_inst->c, 1); + + return error; +} + + cs_error_t cpg_mcast_joined ( cpg_handle_t handle, cpg_guarantee_t guarantee, const struct iovec *iovec, unsigned int iov_len) { int i; cs_error_t error; struct cpg_inst *cpg_inst; struct iovec iov[64]; struct req_lib_cpg_mcast req_lib_cpg_mcast; size_t msg_len = 0; error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } for (i = 0; i < iov_len; i++ ) { msg_len += iovec[i].iov_len; } + if (msg_len > cpg_inst->max_msg_size) { + error = send_fragments(cpg_inst, guarantee, msg_len, iovec, iov_len); + goto error_exit; + } + req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) + msg_len; req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST; req_lib_cpg_mcast.guarantee = guarantee; req_lib_cpg_mcast.msglen = msg_len; iov[0].iov_base = (void *)&req_lib_cpg_mcast; iov[0].iov_len = sizeof (struct req_lib_cpg_mcast); memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec)); qb_ipcc_fc_enable_max_set(cpg_inst->c, 2); error = qb_to_cs_error(qb_ipcc_sendv(cpg_inst->c, iov, iov_len + 1)); qb_ipcc_fc_enable_max_set(cpg_inst->c, 1); +error_exit: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_iteration_initialize( cpg_handle_t handle, cpg_iteration_type_t iteration_type, const struct cpg_name *group, cpg_iteration_handle_t *cpg_iteration_handle) { cs_error_t error; struct iovec iov; struct cpg_inst *cpg_inst; struct cpg_iteration_instance_t *cpg_iteration_instance; struct req_lib_cpg_iterationinitialize req_lib_cpg_iterationinitialize; struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize; if (group && group->length > CPG_MAX_NAME_LENGTH) { return (CS_ERR_NAME_TOO_LONG); } if (cpg_iteration_handle == NULL) { return (CS_ERR_INVALID_PARAM); } if ((iteration_type == CPG_ITERATION_ONE_GROUP && group == NULL) || (iteration_type != CPG_ITERATION_ONE_GROUP && group != NULL)) { return (CS_ERR_INVALID_PARAM); } if (iteration_type != CPG_ITERATION_NAME_ONLY && iteration_type != CPG_ITERATION_ONE_GROUP && iteration_type != CPG_ITERATION_ALL) { return (CS_ERR_INVALID_PARAM); } error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst)); if (error != CS_OK) { return (error); } error = hdb_error_to_cs (hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance_t), cpg_iteration_handle)); if (error != CS_OK) { goto error_put_cpg_db; } error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, *cpg_iteration_handle, (void *)&cpg_iteration_instance)); if (error != CS_OK) { goto error_destroy; } cpg_iteration_instance->conn = cpg_inst->c; list_init (&cpg_iteration_instance->list); req_lib_cpg_iterationinitialize.header.size = sizeof (struct req_lib_cpg_iterationinitialize); req_lib_cpg_iterationinitialize.header.id = MESSAGE_REQ_CPG_ITERATIONINITIALIZE; req_lib_cpg_iterationinitialize.iteration_type = iteration_type; if (group) { marshall_to_mar_cpg_name_t (&req_lib_cpg_iterationinitialize.group_name, group); } iov.iov_base = (void *)&req_lib_cpg_iterationinitialize; iov.iov_len = sizeof (struct req_lib_cpg_iterationinitialize); error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1, &res_lib_cpg_iterationinitialize, sizeof (struct res_lib_cpg_iterationinitialize)); if (error != CS_OK) { goto error_put_destroy; } cpg_iteration_instance->executive_iteration_handle = res_lib_cpg_iterationinitialize.iteration_handle; cpg_iteration_instance->cpg_iteration_handle = *cpg_iteration_handle; list_add (&cpg_iteration_instance->list, &cpg_inst->iteration_list_head); hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle); hdb_handle_put (&cpg_handle_t_db, handle); return (res_lib_cpg_iterationinitialize.header.error); error_put_destroy: hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle); error_destroy: hdb_handle_destroy (&cpg_iteration_handle_t_db, *cpg_iteration_handle); error_put_cpg_db: hdb_handle_put (&cpg_handle_t_db, handle); return (error); } cs_error_t cpg_iteration_next( cpg_iteration_handle_t handle, struct cpg_iteration_description_t *description) { cs_error_t error; struct cpg_iteration_instance_t *cpg_iteration_instance; struct req_lib_cpg_iterationnext req_lib_cpg_iterationnext; struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext; if (description == NULL) { return CS_ERR_INVALID_PARAM; } error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle, (void *)&cpg_iteration_instance)); if (error != CS_OK) { goto error_exit; } req_lib_cpg_iterationnext.header.size = sizeof (struct req_lib_cpg_iterationnext); req_lib_cpg_iterationnext.header.id = MESSAGE_REQ_CPG_ITERATIONNEXT; req_lib_cpg_iterationnext.iteration_handle = cpg_iteration_instance->executive_iteration_handle; error = qb_to_cs_error (qb_ipcc_send (cpg_iteration_instance->conn, &req_lib_cpg_iterationnext, req_lib_cpg_iterationnext.header.size)); if (error != CS_OK) { goto error_put; } error = qb_to_cs_error (qb_ipcc_recv (cpg_iteration_instance->conn, &res_lib_cpg_iterationnext, sizeof(struct res_lib_cpg_iterationnext), -1)); if (error != CS_OK) { goto error_put; } marshall_from_mar_cpg_iteration_description_t( description, &res_lib_cpg_iterationnext.description); error = res_lib_cpg_iterationnext.header.error; error_put: hdb_handle_put (&cpg_iteration_handle_t_db, handle); error_exit: return (error); } cs_error_t cpg_iteration_finalize ( cpg_iteration_handle_t handle) { cs_error_t error; struct iovec iov; struct cpg_iteration_instance_t *cpg_iteration_instance; struct req_lib_cpg_iterationfinalize req_lib_cpg_iterationfinalize; struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize; error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle, (void *)&cpg_iteration_instance)); if (error != CS_OK) { goto error_exit; } req_lib_cpg_iterationfinalize.header.size = sizeof (struct req_lib_cpg_iterationfinalize); req_lib_cpg_iterationfinalize.header.id = MESSAGE_REQ_CPG_ITERATIONFINALIZE; req_lib_cpg_iterationfinalize.iteration_handle = cpg_iteration_instance->executive_iteration_handle; iov.iov_base = (void *)&req_lib_cpg_iterationfinalize; iov.iov_len = sizeof (struct req_lib_cpg_iterationfinalize); error = coroipcc_msg_send_reply_receive (cpg_iteration_instance->conn, &iov, 1, &res_lib_cpg_iterationfinalize, sizeof (struct req_lib_cpg_iterationfinalize)); if (error != CS_OK) { goto error_put; } cpg_iteration_instance_finalize (cpg_iteration_instance); hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle); return (res_lib_cpg_iterationfinalize.header.error); error_put: hdb_handle_put (&cpg_iteration_handle_t_db, handle); error_exit: return (error); } /** @} */ diff --git a/test/Makefile.am b/test/Makefile.am index c19e5068..bb11518c 100644 --- a/test/Makefile.am +++ b/test/Makefile.am @@ -1,72 +1,73 @@ # # Copyright (c) 2009 Red Hat, Inc. # # Authors: Andrew Beekhof # Steven Dake (sdake@redhat.com) # # This software licensed under BSD license, the text of which follows: # # 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. # - Neither the name of the MontaVista Software, Inc. nor the names of its # contributors may 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 OWNER 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. MAINTAINERCLEANFILES = Makefile.in EXTRA_DIST = ploadstart.sh -noinst_PROGRAMS = cpgverify testcpg testcpg2 cpgbench \ +noinst_PROGRAMS = cpgverify testcpg testcpg2 cpgbench cpghum \ testquorum testvotequorum1 testvotequorum2 \ stress_cpgfdget stress_cpgcontext cpgbound testsam \ testcpgzc cpgbenchzc testzcgc stress_cpgzc noinst_SCRIPTS = ploadstart testcpg_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la testcpg2_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la testcpgzc_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la testzcgc_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la stress_cpgzc_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la stress_cpgfdget_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la stress_cpgcontext_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la +cpghum_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la -lz testquorum_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libquorum.la testvotequorum1_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libvotequorum.la testvotequorum2_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libvotequorum.la cpgverify_CPPFLAGS = $(nss_CFLAGS) cpgverify_LDADD = $(LIBQB_LIBS) $(nss_LIBS) $(top_builddir)/lib/libcpg.la cpgbound_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la cpgbench_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la cpgbenchzc_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libcpg.la testsam_LDADD = $(LIBQB_LIBS) $(top_builddir)/lib/libsam.la ploadstart: ploadstart.sh sed -e 's#@''BASHPATH@#${BASHPATH}#g' $< > $@ chmod 755 $@ LINT_FILES1:=$(filter-out sa_error.c, $(wildcard *.c)) LINT_FILES:=$(filter-out testparse.c, $(LINT_FILES1)) lint: -for f in $(LINT_FILES) ; do echo Splint $$f ; splint $(LINT_FLAGS) $(CPPFLAGS) $(CFLAGS) $$f ; done clean-local: rm -f ploadstart