Allow more children per entity

Sometimes application overflow the children counters in the handle
flags. That leads to a bad crash, which is a bug, but it is also less
than ideal than one can't make a great many child entities.

With the way the bits are mapped, one can have 16k children of a single
entity and 4k concurrent calls on a single entity. This commit changes
that in two ways:

* On 32-bit platforms, it shifts the division of bits, allowing up to
  64k children and up to 1k concurrent calls on a single entity. It is
  clearly still possible to run into the same bug, and even slightly
  more likely to do so in progress with very many threads, but this
  seems like a minor improvement nonetheless.

* On 64-bit platforms, it changes the representation to an 64-bit
  integer (it simply uses an uintptr_t), which gives plenty of room in
  the field to substantially increase the width of the counters.

Just widening the counters is not an elegant solution, and 64-bits here
is a bit wasteful. Still, this should fix the problem at least in
practice.

Signed-off-by: Erik Boasson <[email protected]>

Author: eboasson

src/core/ddsc/src/dds__handles.h

@@ -14,6 +14,7 @@
 #include "dds/ddsrt/time.h"
 #include "dds/ddsrt/retcode.h"
 #include "dds/ddsrt/atomics.h"
+#include "dds/ddsrt/arch.h"
 #include "dds/dds.h"
 
 #if defined (__cplusplus)
@@ -63,16 +64,29 @@ typedef int32_t dds_handle_t;
 
 /* Closing & closed can be combined, but having two gives a means for enforcing
    that close() be called first, then close_wait(), and then delete(). */
+#if DDSRT_64BIT // Set for WIN64, _LP64 => sizeof(void*)=8 => uintptr_t=8 is a reasonable assumption
+
+#define HDL_FLAG_CLOSING         UINT64_C(0x8000000000000000)
+#define HDL_FLAG_DELETE_DEFERRED UINT64_C(0x4000000000000000)
+#define HDL_FLAG_PENDING         UINT64_C(0x2000000000000000)
+#define HDL_FLAG_IMPLICIT        UINT64_C(0x1000000000000000)
+#define HDL_FLAG_ALLOW_CHILDREN  UINT64_C(0x0800000000000000) /* refc counts children */
+#define HDL_FLAG_NO_USER_ACCESS  UINT64_C(0x0400000000000000)
+
+#else
+
 #define HDL_FLAG_CLOSING         (0x80000000u)
 #define HDL_FLAG_DELETE_DEFERRED (0x40000000u)
 #define HDL_FLAG_PENDING         (0x20000000u)
 #define HDL_FLAG_IMPLICIT        (0x10000000u)
 #define HDL_FLAG_ALLOW_CHILDREN  (0x08000000u) /* refc counts children */
 #define HDL_FLAG_NO_USER_ACCESS  (0x04000000u)
 
+#endif
+
 struct dds_handle_link {
   dds_handle_t hdl;
-  ddsrt_atomic_uint32_t cnt_flags;
+  ddsrt_atomic_uintptr_t cnt_flags;
 };
 
 /**
@@ -206,7 +220,7 @@ bool dds_handle_unpin_and_drop_ref (struct dds_handle_link *link);
  * possible.
  */
 inline bool dds_handle_is_closed (struct dds_handle_link *link) {
-  return (ddsrt_atomic_ld32 (&link->cnt_flags) & HDL_FLAG_CLOSING) != 0;
+  return (ddsrt_atomic_ldptr (&link->cnt_flags) & HDL_FLAG_CLOSING) != 0;
 }
 
 /** @component handles */

src/core/ddsc/src/dds_entity.c

@@ -307,7 +307,7 @@ void dds_entity_init_complete (dds_entity *entity)
 void dds_entity_register_child (dds_entity *parent, dds_entity *child)
 {
   /* parent must be tracking children in its refc, or children can't be added */
-  assert (ddsrt_atomic_ld32 (&parent->m_hdllink.cnt_flags) & HDL_FLAG_ALLOW_CHILDREN);
+  assert (ddsrt_atomic_ldptr (&parent->m_hdllink.cnt_flags) & HDL_FLAG_ALLOW_CHILDREN);
   assert (child->m_iid != 0);
   assert (ddsrt_avl_lookup (&dds_entity_children_td, &parent->m_children, &child->m_iid) == NULL);
   ddsrt_avl_insert (&dds_entity_children_td, &parent->m_children, child);

src/core/ddsc/src/dds_handles.c

@@ -19,10 +19,21 @@
 #include "dds__handles.h"
 #include "dds__types.h"
 
-#define HDL_REFCOUNT_MASK  (0x03fff000u)
-#define HDL_REFCOUNT_UNIT  (0x00001000u)
-#define HDL_REFCOUNT_SHIFT 12
-#define HDL_PINCOUNT_MASK  (0x00000fffu)
+#if DDSRT_64BIT // see dds__handles.h for reasoning
+
+#define HDL_REFCOUNT_MASK  UINT64_C(0x0003ffffff000000)
+#define HDL_REFCOUNT_UNIT  UINT64_C(0x0000000001000000)
+#define HDL_REFCOUNT_SHIFT 24
+#define HDL_PINCOUNT_MASK  UINT64_C(0x0000000000ffffff)
+
+#else
+
+#define HDL_REFCOUNT_MASK  (0x03fffc00u)
+#define HDL_REFCOUNT_UNIT  (0x00000400u)
+#define HDL_REFCOUNT_SHIFT 10
+#define HDL_PINCOUNT_MASK  (0x000003ffu)
+
+#endif
 
 /*
 "regular" entities other than topics:
@@ -101,9 +112,10 @@ void dds_handle_server_fini (void)
     struct ddsrt_hh_iter it;
     for (struct dds_handle_link *link = ddsrt_hh_iter_first (handles.ht, &it); link != NULL; link = ddsrt_hh_iter_next (&it))
     {
-      uint32_t cf = ddsrt_atomic_ld32 (&link->cnt_flags);
-      DDS_ERROR ("handle %"PRId32" pin %"PRIu32" refc %"PRIu32"%s%s%s\n", link->hdl,
-                 cf & HDL_PINCOUNT_MASK, (cf & HDL_REFCOUNT_MASK) >> HDL_REFCOUNT_SHIFT,
+      uintptr_t cf = ddsrt_atomic_ldptr (&link->cnt_flags);
+      DDS_ERROR ("handle %"PRId32" pin %"PRIuPTR" refc %"PRIuPTR"%s%s%s\n", link->hdl,
+                 (uintptr_t) (cf & HDL_PINCOUNT_MASK),
+                 (uintptr_t) ((cf & HDL_REFCOUNT_MASK) >> HDL_REFCOUNT_SHIFT),
                  cf & HDL_FLAG_PENDING ? " pending" : "",
                  cf & HDL_FLAG_CLOSING ? " closing" : "",
                  cf & HDL_FLAG_DELETE_DEFERRED ? " delete-deferred" : "");
@@ -119,11 +131,11 @@ void dds_handle_server_fini (void)
 
 static dds_handle_t dds_handle_create_int (struct dds_handle_link *link, bool implicit, bool refc_counts_children, bool user_access)
 {
-  uint32_t flags = HDL_FLAG_PENDING;
+  uintptr_t flags = HDL_FLAG_PENDING;
   flags |= implicit ? HDL_FLAG_IMPLICIT : HDL_REFCOUNT_UNIT;
   flags |= refc_counts_children ? HDL_FLAG_ALLOW_CHILDREN : 0;
   flags |= user_access ? 0 : HDL_FLAG_NO_USER_ACCESS;
-  ddsrt_atomic_st32 (&link->cnt_flags, flags | 1u);
+  ddsrt_atomic_stptr (&link->cnt_flags, flags | 1u);
   do {
     do {
       link->hdl = (int32_t) (ddsrt_random () & INT32_MAX);
@@ -165,7 +177,7 @@ dds_return_t dds_handle_register_special (struct dds_handle_link *link, bool imp
   else
   {
     handles.count++;
-    ddsrt_atomic_st32 (&link->cnt_flags, HDL_FLAG_PENDING | (implicit ? HDL_FLAG_IMPLICIT : HDL_REFCOUNT_UNIT) | (allow_children ? HDL_FLAG_ALLOW_CHILDREN : 0) | 1u);
+    ddsrt_atomic_stptr (&link->cnt_flags, HDL_FLAG_PENDING | (implicit ? HDL_FLAG_IMPLICIT : HDL_REFCOUNT_UNIT) | (allow_children ? HDL_FLAG_ALLOW_CHILDREN : 0) | 1u);
     link->hdl = handle;
     if (ddsrt_hh_add (handles.ht, link))
       ret = handle;
@@ -180,21 +192,21 @@ dds_return_t dds_handle_register_special (struct dds_handle_link *link, bool imp
 void dds_handle_unpend (struct dds_handle_link *link)
 {
 #ifndef NDEBUG
-  uint32_t cf = ddsrt_atomic_ld32 (&link->cnt_flags);
+  uintptr_t cf = ddsrt_atomic_ldptr (&link->cnt_flags);
   assert ((cf & HDL_FLAG_PENDING));
   assert (!(cf & HDL_FLAG_DELETE_DEFERRED));
   assert (!(cf & HDL_FLAG_CLOSING));
   assert ((cf & HDL_REFCOUNT_MASK) >= HDL_REFCOUNT_UNIT || (cf & HDL_FLAG_IMPLICIT));
   assert ((cf & HDL_PINCOUNT_MASK) >= 1u);
 #endif
-  ddsrt_atomic_and32 (&link->cnt_flags, ~HDL_FLAG_PENDING);
+  ddsrt_atomic_andptr (&link->cnt_flags, ~HDL_FLAG_PENDING);
   dds_handle_unpin (link);
 }
 
 int32_t dds_handle_delete (struct dds_handle_link *link)
 {
 #ifndef NDEBUG
-  uint32_t cf = ddsrt_atomic_ld32 (&link->cnt_flags);
+  uintptr_t cf = ddsrt_atomic_ldptr (&link->cnt_flags);
   if (!(cf & HDL_FLAG_PENDING))
   {
     assert (cf & HDL_FLAG_CLOSING);
@@ -210,7 +222,7 @@ int32_t dds_handle_delete (struct dds_handle_link *link)
   return DDS_RETCODE_OK;
 }
 
-static int32_t dds_handle_pin_int (dds_handle_t hdl, uint32_t delta, bool from_user, struct dds_handle_link **link)
+static int32_t dds_handle_pin_int (dds_handle_t hdl, uintptr_t delta, bool from_user, struct dds_handle_link **link)
 {
   struct dds_handle_link dummy = { .hdl = hdl };
   int32_t rc;
@@ -231,12 +243,12 @@ static int32_t dds_handle_pin_int (dds_handle_t hdl, uint32_t delta, bool from_u
     rc = DDS_RETCODE_BAD_PARAMETER;
   else
   {
-    uint32_t cf;
+    uintptr_t cf;
     /* Assume success; bail out if the object turns out to be in the process of
        being deleted */
     rc = DDS_RETCODE_OK;
     do {
-      cf = ddsrt_atomic_ld32 (&(*link)->cnt_flags);
+      cf = ddsrt_atomic_ldptr (&(*link)->cnt_flags);
       if (cf & (HDL_FLAG_CLOSING | HDL_FLAG_PENDING | HDL_FLAG_NO_USER_ACCESS))
       {
         if (cf & (HDL_FLAG_CLOSING | HDL_FLAG_PENDING))
@@ -250,7 +262,7 @@ static int32_t dds_handle_pin_int (dds_handle_t hdl, uint32_t delta, bool from_u
           break;
         }
       }
-    } while (!ddsrt_atomic_cas32 (&(*link)->cnt_flags, cf, cf + delta));
+    } while (!ddsrt_atomic_casptr (&(*link)->cnt_flags, cf, cf + delta));
   }
   ddsrt_mutex_unlock (&handles.lock);
   return rc;
@@ -287,11 +299,11 @@ int32_t dds_handle_pin_for_delete (dds_handle_t hdl, bool explicit, bool from_us
     rc = DDS_RETCODE_BAD_PARAMETER;
   else
   {
-    uint32_t cf, cf1;
+    uintptr_t cf, cf1;
     /* Assume success; bail out if the object turns out to be in the process of
        being deleted */
     do {
-      cf = ddsrt_atomic_ld32 (&(*link)->cnt_flags);
+      cf = ddsrt_atomic_ldptr (&(*link)->cnt_flags);
 
       if (from_user && (cf & (HDL_FLAG_NO_USER_ACCESS)))
       {
@@ -384,7 +396,7 @@ int32_t dds_handle_pin_for_delete (dds_handle_t hdl, bool explicit, bool from_us
       }
 
       rc = ((cf1 & HDL_REFCOUNT_MASK) == 0 || (cf1 & HDL_FLAG_ALLOW_CHILDREN)) ? DDS_RETCODE_OK : DDS_RETCODE_TRY_AGAIN;
-    } while (!ddsrt_atomic_cas32 (&(*link)->cnt_flags, cf, cf1));
+    } while (!ddsrt_atomic_casptr (&(*link)->cnt_flags, cf, cf1));
   }
   ddsrt_mutex_unlock (&handles.lock);
   return rc;
@@ -394,9 +406,9 @@ bool dds_handle_drop_childref_and_pin (struct dds_handle_link *link, bool may_de
 {
   bool del_parent = false;
   ddsrt_mutex_lock (&handles.lock);
-  uint32_t cf, cf1;
+  uintptr_t cf, cf1;
   do {
-    cf = ddsrt_atomic_ld32 (&link->cnt_flags);
+    cf = ddsrt_atomic_ldptr (&link->cnt_flags);
 
     if (cf & (HDL_FLAG_CLOSING | HDL_FLAG_PENDING))
     {
@@ -430,7 +442,7 @@ bool dds_handle_drop_childref_and_pin (struct dds_handle_link *link, bool may_de
         del_parent = false;
       }
     }
-  } while (!ddsrt_atomic_cas32 (&link->cnt_flags, cf, cf1));
+  } while (!ddsrt_atomic_casptr (&link->cnt_flags, cf, cf1));
   ddsrt_mutex_unlock (&handles.lock);
   return del_parent;
 }
@@ -442,21 +454,21 @@ int32_t dds_handle_pin_and_ref_with_origin (dds_handle_t hdl, bool from_user, st
 
 void dds_handle_repin (struct dds_handle_link *link)
 {
-  uint32_t x = ddsrt_atomic_inc32_nv (&link->cnt_flags);
+  uintptr_t x = ddsrt_atomic_incptr_nv (&link->cnt_flags);
   (void) x;
 }
 
 void dds_handle_unpin (struct dds_handle_link *link)
 {
 #ifndef NDEBUG
-  uint32_t cf = ddsrt_atomic_ld32 (&link->cnt_flags);
+  uintptr_t cf = ddsrt_atomic_ldptr (&link->cnt_flags);
   if (cf & HDL_FLAG_CLOSING)
     assert ((cf & HDL_PINCOUNT_MASK) > 1u);
   else
     assert ((cf & HDL_PINCOUNT_MASK) >= 1u);
 #endif
   ddsrt_mutex_lock (&handles.lock);
-  if ((ddsrt_atomic_dec32_nv (&link->cnt_flags) & (HDL_FLAG_CLOSING | HDL_PINCOUNT_MASK)) == (HDL_FLAG_CLOSING | 1u))
+  if ((ddsrt_atomic_decptr_nv (&link->cnt_flags) & (HDL_FLAG_CLOSING | HDL_PINCOUNT_MASK)) == (HDL_FLAG_CLOSING | 1u))
   {
     ddsrt_cond_broadcast (&handles.cond);
   }
@@ -465,17 +477,17 @@ void dds_handle_unpin (struct dds_handle_link *link)
 
 void dds_handle_add_ref (struct dds_handle_link *link)
 {
-  ddsrt_atomic_add32 (&link->cnt_flags, HDL_REFCOUNT_UNIT);
+  ddsrt_atomic_addptr (&link->cnt_flags, HDL_REFCOUNT_UNIT);
 }
 
 bool dds_handle_drop_ref (struct dds_handle_link *link)
 {
-  uint32_t old, new;
+  uintptr_t old, new;
   do {
-    old = ddsrt_atomic_ld32 (&link->cnt_flags);
+    old = ddsrt_atomic_ldptr (&link->cnt_flags);
     assert ((old & HDL_REFCOUNT_MASK) > 0);
     new = old - HDL_REFCOUNT_UNIT;
-  } while (!ddsrt_atomic_cas32 (&link->cnt_flags, old, new));
+  } while (!ddsrt_atomic_casptr (&link->cnt_flags, old, new));
   ddsrt_mutex_lock (&handles.lock);
   if ((new & (HDL_FLAG_CLOSING | HDL_PINCOUNT_MASK)) == (HDL_FLAG_CLOSING | 1u))
   {
@@ -487,13 +499,13 @@ bool dds_handle_drop_ref (struct dds_handle_link *link)
 
 bool dds_handle_unpin_and_drop_ref (struct dds_handle_link *link)
 {
-  uint32_t old, new;
+  uintptr_t old, new;
   do {
-    old = ddsrt_atomic_ld32 (&link->cnt_flags);
+    old = ddsrt_atomic_ldptr (&link->cnt_flags);
     assert ((old & HDL_REFCOUNT_MASK) > 0);
     assert ((old & HDL_PINCOUNT_MASK) > 0);
     new = old - HDL_REFCOUNT_UNIT - 1u;
-  } while (!ddsrt_atomic_cas32 (&link->cnt_flags, old, new));
+  } while (!ddsrt_atomic_casptr (&link->cnt_flags, old, new));
   ddsrt_mutex_lock (&handles.lock);
   if ((new & (HDL_FLAG_CLOSING | HDL_PINCOUNT_MASK)) == (HDL_FLAG_CLOSING | 1u))
   {
@@ -505,27 +517,27 @@ bool dds_handle_unpin_and_drop_ref (struct dds_handle_link *link)
 
 bool dds_handle_close (struct dds_handle_link *link)
 {
-  uint32_t old = ddsrt_atomic_or32_ov (&link->cnt_flags, HDL_FLAG_CLOSING);
+  uintptr_t old = ddsrt_atomic_orptr_ov (&link->cnt_flags, HDL_FLAG_CLOSING);
   return (old & HDL_REFCOUNT_MASK) == 0;
 }
 
 void dds_handle_close_wait (struct dds_handle_link *link)
 {
 #ifndef NDEBUG
-  uint32_t cf = ddsrt_atomic_ld32 (&link->cnt_flags);
+  uintptr_t cf = ddsrt_atomic_ldptr (&link->cnt_flags);
   assert ((cf & HDL_FLAG_CLOSING));
   assert ((cf & HDL_PINCOUNT_MASK) >= 1u);
 #endif
   ddsrt_mutex_lock (&handles.lock);
-  while ((ddsrt_atomic_ld32 (&link->cnt_flags) & HDL_PINCOUNT_MASK) != 1u)
+  while ((ddsrt_atomic_ldptr (&link->cnt_flags) & HDL_PINCOUNT_MASK) != 1u)
     ddsrt_cond_wait (&handles.cond, &handles.lock);
   /* only one thread may call close_wait on a given handle */
   ddsrt_mutex_unlock (&handles.lock);
 }
 
 bool dds_handle_is_not_refd (struct dds_handle_link *link)
 {
-  return ((ddsrt_atomic_ld32 (&link->cnt_flags) & HDL_REFCOUNT_MASK) == 0);
+  return ((ddsrt_atomic_ldptr (&link->cnt_flags) & HDL_REFCOUNT_MASK) == 0);
 }
 
 extern inline bool dds_handle_is_closed (struct dds_handle_link *link);