Fixing #800: Add possibility to disable array optimizations during se…

…rialization

docs/manual/config_defaults.qbk

@@ -241,6 +241,7 @@ sections, one for each component.
     max_connections = ${HPX_PARCEL_MAX_CONNECTIONS:<hpx_parcel_max_connections>}
     max_connections_per_locality = ${HPX_PARCEL_MAX_CONNECTIONS_PER_LOCALITY:<hpx_parcel_max_connections_per_locality>}
     max_message_size = ${HPX_PARCEL_MAX_MESSAGE_SIZE:<hpx_parcel_max_message_size>}
+    array_optimization = ${HPX_PARCEL_ARRAY_OPTIMIZATION:1}
 ``
 [c++]
 
@@ -281,6 +282,10 @@ sections, one for each component.
       transferrable through the parcel layer.]
      [static information, the default depends on the compile time preprocessor
       constant `HPX_PARCEL_MAX_MESSAGE_SIZE` (`1000000000`) bytes.]]
+    [[`hpx.parcel.array_optimization`]
+     [This property defines whether this locality is allowed to utilize array
+      optimizations during serialization of parcel data.]
+     [static information, the default is `1` (on).]]
 ]
 
 [teletype]

hpx/config.hpp

@@ -71,6 +71,11 @@
 #  define FUSION_MAX_VECTOR_SIZE 20
 #endif
 
+// make sure boost::result_of is adjusted appropriately as well
+#if HPX_LIMIT > 5 && !defined(BOOST_RESULT_OF_NUM_ARGS)
+#  define BOOST_RESULT_OF_NUM_ARGS 20
+#endif
+
 ///////////////////////////////////////////////////////////////////////////////
 // We currently do not support more than 20 arguments (ask if you need more)
 #if !defined(HPX_MAX_LIMIT)
@@ -645,10 +650,10 @@
 #define HPX_AGAS_LOCALITY_NS_LSB                     0x0000000000000004ULL
 
 #if defined(HPX_HAVE_SODIUM)
-#  define HPX_ROOT_CERTIFICATE_AUTHORITY_MSB           0x0000000100000001ULL
-#  define HPX_ROOT_CERTIFICATE_AUTHORITY_LSB           0x0000000000000005ULL
-#  define HPX_SUBORDINATE_CERTIFICATE_AUTHORITY_MSB    0x0000000000000001ULL      // this is made locality specific
-#  define HPX_SUBORDINATE_CERTIFICATE_AUTHORITY_LSB    0x0000000000000006ULL
+#  define HPX_ROOT_CERTIFICATE_AUTHORITY_MSB         0x0000000100000001ULL
+#  define HPX_ROOT_CERTIFICATE_AUTHORITY_LSB         0x0000000000000005ULL
+#  define HPX_SUBORDINATE_CERTIFICATE_AUTHORITY_MSB  0x0000000000000001ULL      // this is made locality specific
+#  define HPX_SUBORDINATE_CERTIFICATE_AUTHORITY_LSB  0x0000000000000006ULL
 #endif
 
 #if !defined(HPX_NO_DEPRECATED)

hpx/util/portable_binary_iarchive.hpp

@@ -120,7 +120,7 @@ class HPX_SERIALIZATION_EXPORT portable_binary_iarchive :
     friend class boost::archive::load_access;
 protected:
 #endif
-    unsigned int m_flags;
+    boost::uint32_t m_flags;
     HPX_ALWAYS_EXPORT void
     load_impl(boost::int64_t& l, char maxsize);
 
@@ -266,7 +266,7 @@ class HPX_SERIALIZATION_EXPORT portable_binary_iarchive :
         init(flags);
     }
 
-    unsigned int flags() const
+    boost::uint32_t flags() const
     {
         return m_flags;
     }

hpx/util/portable_binary_oarchive.hpp

@@ -118,7 +118,7 @@ class HPX_SERIALIZATION_EXPORT portable_binary_oarchive :
     friend class boost::archive::save_access;
 protected:
 #endif
-    unsigned int m_flags;
+    boost::uint32_t m_flags;
     HPX_ALWAYS_EXPORT void
     save_impl(const boost::int64_t l, const char maxsize);
 
@@ -227,12 +227,12 @@ class HPX_SERIALIZATION_EXPORT portable_binary_oarchive :
     portable_binary_oarchive(Container& buffer, binary_filter* filter = 0, unsigned flags = 0)
       : primitive_base_t(buffer, flags),
         archive_base_t(flags),
-        m_flags(flags & (enable_compression | endian_big | endian_little))
+        m_flags(flags & (enable_compression | endian_big | endian_little | disable_array_optimization))
     {
         init(filter, flags);
     }
 
-    unsigned int flags() const
+    boost::uint32_t flags() const
     {
         return m_flags;
     }

src/runtime/parcelset/ibverbs/parcelport_connection_ibverbs.cpp

@@ -49,15 +49,20 @@ namespace hpx { namespace parcelset { namespace ibverbs
             BOOST_ASSERT(endian_out =="little" || endian_out == "big");
         }
 
+        std::string array_optimization =
+            get_config_entry("hpx.parcel.array_optimization", "1");
+        if (boost::lexical_cast<int>(array_optimization) == 0)
+            archive_flags_ |= util::disable_array_optimization;
+
         boost::system::error_code ec;
         std::string buffer_size_str = get_config_entry("hpx.parcel.ibverbs.buffer_size", "4096");
-        
+
         std::size_t buffer_size = boost::lexical_cast<std::size_t>(buffer_size_str);
         char * mr_buffer = context_.set_buffer_size(buffer_size, ec);
 
         out_buffer_.set_mr_buffer(mr_buffer, buffer_size);
     }
-    
+
     ///////////////////////////////////////////////////////////////////////////
     void parcelport_connection::set_parcel(std::vector<parcel> const& pv)
     {

src/runtime/parcelset/ibverbs/parcelport_ibverbs.cpp

@@ -549,6 +549,10 @@ namespace hpx { namespace parcelset { namespace ibverbs
         boost::uint64_t inbound_data_size,
         performance_counters::parcels::data_point receive_data)
     {
+        unsigned archive_flags = boost::archive::no_header;
+        if (!pp.allow_array_optimizations())
+            archive_flags |= util::disable_array_optimization;
+
         // protect from un-handled exceptions bubbling up
         try {
             try {
@@ -559,7 +563,7 @@ namespace hpx { namespace parcelset { namespace ibverbs
                 {
                     // De-serialize the parcel data
                     util::portable_binary_iarchive archive(parcel_data,
-                        parcel_data.size(), boost::archive::no_header);
+                        parcel_data.size(), archive_flags);
 
                     std::size_t parcel_count = 0;
 

src/runtime/parcelset/parcelport.cpp

@@ -139,7 +139,7 @@ namespace hpx { namespace parcelset
         allow_array_optimizations_(true)
     {
         std::string array_optimization =
-            get_config_entry("hpx.parcel.array_optimization", "1");
+            ini.get_entry("hpx.parcel.array_optimization", "1");
         if (boost::lexical_cast<int>(array_optimization) == 0)
             allow_array_optimizations_ = false;
     }

src/util/portable_binary_iarchive.cpp

@@ -142,9 +142,9 @@ void portable_binary_iarchive::init(unsigned int flags)
 #endif
     }
 
-    unsigned char x;
+    boost::uint16_t x;
     load(x);
-    m_flags = static_cast<unsigned int>(x << CHAR_BIT);
+    m_flags = static_cast<boost::uint32_t>(x << CHAR_BIT);
 
     // handle filter and compression in the archive separately
     bool has_filter = false;

src/util/portable_binary_oarchive.cpp

@@ -114,7 +114,7 @@ void portable_binary_oarchive::init(util::binary_filter* filter, unsigned int fl
         *this << v;
     }
 
-    save(static_cast<unsigned char>(m_flags >> CHAR_BIT));
+    save(static_cast<boost::uint16_t>(m_flags >> CHAR_BIT));
 
     // handle filter and compression in the archive separately
     bool has_filter = filter ? true : false;