Merge pull request ceph#59016 from Svelar/wip-revert-rgw

Revert "reef: rgw/amqp: lock erase and create connection before emplace"

src/rgw/rgw_amqp.cc

@@ -535,12 +535,9 @@ typedef boost::lockfree::queue<message_wrapper_t*, boost::lockfree::fixed_sized<
           continue;
 
 #define ERASE_AND_CONTINUE(IT,CONTAINER) \
-          { \
-            std::lock_guard lock(connections_lock); \
-            IT=CONTAINER.erase(IT); \
-            --connection_count; \
-            continue; \
-          }
+          IT=CONTAINER.erase(IT); \
+          --connection_count; \
+          continue;
 
 class Manager {
 public:
@@ -886,14 +883,12 @@ class Manager {
     }
     // if error occurred during creation the creation will be retried in the main thread
     ++connection_count;
-    auto conn = std::make_unique<connection_t>(cct, info, verify_ssl, ca_location);
-    if (new_state(conn.get(), tmp_id)) {
-        ldout(cct, 10) << "AMQP connect: new connection is created. Total connections: " << connection_count << dendl;
-    } else {
-        ldout(cct, 1) << "AMQP connect: new connection '" << to_string(tmp_id) << "' is created. but state creation failed (will retry). error: " <<
-            status_to_string(conn->status) << " (" << conn->reply_code << ")"  << dendl;
+    auto conn = connections.emplace(tmp_id, std::make_unique<connection_t>(cct, info, verify_ssl, ca_location)).first->second.get();
+    ldout(cct, 10) << "AMQP connect: new connection is created. Total connections: " << connection_count << dendl;
+    if (!new_state(conn, tmp_id)) {
+      ldout(cct, 1) << "AMQP connect: new connection '" << to_string(tmp_id) << "' is created. but state creation failed (will retry). error: " <<
+        status_to_string(conn->status) << " (" << conn->reply_code << ")"  << dendl;
     }
-    connections.emplace(tmp_id, std::move(conn));
     id = std::move(tmp_id);
     return true;
   }