Never exceed persistent connection limits

By doing the persistent connection count incrementation with CAS, value
can be guaranteed to never exceed the configured limit.

maxutils/maxbase/include/maxbase/atomic.hh

@@ -70,5 +70,55 @@ void store(T* t, R v, int mode = SEQ_CST)
 {
     __atomic_store_n(t, v, mode);
 }
+
+/**
+ * Perform atomic compare and exchange operation
+ *
+ * @param ptr           Variable where the value is stored
+ * @param expected      Expected value of the variable
+ * @param desired       The desired new value of the variable
+ * @param success_model On success, this memory model is used
+ * @param fail_model    On failure, this memory model is used
+ *
+ * @return True if value was exchanged, false if exchange failed
+ */
+template<class T>
+bool compare_exchange(T* ptr, T* expected, T desired, int success_model = ACQ_REL, int fail_model = ACQUIRE)
+{
+    return __atomic_compare_exchange_n(ptr, expected, desired, true, success_model, fail_model);
+}
+
+/**
+ * Add to a value if it doesn't exceed a limit
+ *
+ * If the value of `ptr` + `value` is less than or equal to `limit`, the value is atomically added.
+ *
+ * @param ptr   Pointer to value to add to
+ * @param value Value to add
+ * @param limit Upper limit that is not exceeded
+ *
+ * @return True if value was modified, false if the addition failed.
+ */
+template<class T>
+bool add_limited(T* ptr, T value, T limit)
+{
+    T expected;
+    T next_value;
+
+    do
+    {
+        expected = mxb::atomic::load(ptr, mxb::atomic::ACQUIRE);
+
+        if (limit < expected + value)
+        {
+            return false;
+        }
+
+        next_value = expected + value;
+    }
+    while (!mxb::atomic::compare_exchange(ptr, &expected, next_value));
+
+    return true;
+}
 }
 }

server/core/dcb.cc

@@ -1319,9 +1319,13 @@ static bool dcb_maybe_add_persistent(DCB* dcb)
         && (dcb->server->status & SERVER_RUNNING)
         && !dcb->dcb_errhandle_called
         && !(dcb->flags & DCBF_HUNG)
-        && dcb_persistent_clean_count(dcb, owner->id(), false) < dcb->server->persistpoolmax
+        && dcb_persistent_clean_count(dcb, owner->id(), false) < dcb->server->persistpoolmax)
-        && mxb::atomic::load(&dcb->server->stats.n_persistent) < dcb->server->persistpoolmax)
     {
+        if (!mxb::atomic::add_limited(&dcb->server->stats.n_persistent, 1, (int)dcb->server->persistpoolmax))
+        {
+            return false;
+        }
+
         DCB_CALLBACK* loopcallback;
         MXS_DEBUG("Adding DCB to persistent pool, user %s.", dcb->user);
         dcb->was_persistent = false;
@@ -1357,7 +1361,6 @@ static bool dcb_maybe_add_persistent(DCB* dcb)
 
         dcb->nextpersistent = dcb->server->persistent[owner->id()];
         dcb->server->persistent[owner->id()] = dcb;
-        mxb::atomic::add(&dcb->server->stats.n_persistent, 1);
         MXB_AT_DEBUG(int rc = ) mxb::atomic::add(&dcb->server->stats.n_current, -1, mxb::atomic::RELAXED);
         mxb_assert(rc > 0);
         return true;