Add support to replace the AccessPoint in ProxyDestination

Summary:
Add support to replace the AccessPoint in ProxyDestination

1. Use ProxyMessage to send message to Proxy thread to do the actual change of access point in Proxy thread context and not in config thread context. The Proxy object does the actual replacement of the access point by finding the ProxyDestination in the ProxyDestinationMap using the temporary old access point constructed from the IP address that is replaced.

2. Put the new destination in TKO mode (because the new host won't be ready anyways and so pro-actively put the destination in TKO mode)

3. replace shared_ptr<AccessPoint> in ProxyDestination with folly::atomic_shared_ptr<AccessPoint>

Description of the AP replacement logic:
```
caller calls Proxy->replaceAccessPoint() with temp old AP constructed from old IP address/port
            send a message to Proxy thread to find and replace old AP
```

Proxy thread upon receiving the replace AP message
```
calls processReplaceMessage( tmpOldAP, newAP)
    which calls ProxyDestinationMap::replace(tmpOldAP, newAP)
       This function uses tmpOldAP to find the realOldAP in the Proxy DestinationMap
       If it is found, it is erased first because the MAP used AP internally to identify the destination
       Then replaces the old AP with new AP
       and inserts the destination back into the MAP which internally uses new AP to correctly place it in the MAP
       Finally returns the old AP so that it map be used by the caller
```

Reviewed By: stuclar

Differential Revision: D25416686

fbshipit-source-id: 7ad38e21dd60811f75d56903c0d6a029363c459d

Author: Krishna Kondaka

mcrouter/Proxy-inl.h

@@ -314,6 +314,12 @@ void Proxy<RouterInfo>::messageReady(ProxyMessage::Type t, void* data) {
       delete oldConfig;
     } break;
 
+    case ProxyMessage::Type::REPLACE_AP: {
+      auto rep = reinterpret_cast<replace_ap_t*>(data);
+      processReplaceMessage(rep);
+      rep->baton.post();
+    } break;
+
     case ProxyMessage::Type::SHUTDOWN:
       /*
        * No-op. We just wanted to wake this event base up so that
@@ -422,6 +428,41 @@ bool Proxy<RouterInfo>::messageQueueFull() const noexcept {
   return messageQueue_->isFull();
 }
 
+template <class RouterInfo>
+void Proxy<RouterInfo>::processReplaceMessage(replace_ap_t* rep) {
+  rep->replacedAccessPoint = nullptr;
+  if (rep->oldAccessPoint.getProtocol() == mc_thrift_protocol) {
+    rep->replacedAccessPoint =
+        destinationMap_->replace<ThriftTransport<RouterInfo>>(
+            rep->oldAccessPoint,
+            rep->newAccessPoint,
+            std::chrono::milliseconds(0));
+  } else if (rep->oldAccessPoint.getProtocol() == mc_caret_protocol) {
+    rep->replacedAccessPoint = destinationMap_->replace<AsyncMcClient>(
+        rep->oldAccessPoint, rep->newAccessPoint, std::chrono::milliseconds(0));
+  }
+}
+
+template <class RouterInfo>
+std::shared_ptr<const AccessPoint> Proxy<RouterInfo>::replaceAP(
+    const AccessPoint& oldAP,
+    std::shared_ptr<const AccessPoint> newAP) {
+  // Send message to replace AP in the ProxyDestination
+  try {
+    auto repMsg = std::make_unique<replace_ap_t>(oldAP, newAP);
+    sendMessage(ProxyMessage::Type::REPLACE_AP, repMsg.get());
+    // wait for the response
+    repMsg->baton.wait();
+    auto replacedAP = repMsg->replacedAccessPoint;
+    VLOG(2) << ((replacedAP != nullptr) ? "SUCCESS " : "FAILED ")
+            << " Replacing oldAP = " << oldAP.toString()
+            << " newAP = " << newAP->toString();
+    return replacedAP;
+  } catch (...) {
+    return nullptr;
+  }
+}
+
 } // namespace mcrouter
 } // namespace memcache
 } // namespace facebook

mcrouter/Proxy.h

@@ -33,7 +33,9 @@
 #include "mcrouter/lib/carbon/Keys.h"
 #include "mcrouter/lib/mc/msg.h"
 #include "mcrouter/lib/mc/protocol.h"
+#include "mcrouter/lib/network/AsyncMcClient.h"
 #include "mcrouter/lib/network/CarbonMessageList.h"
+#include "mcrouter/lib/network/ThriftTransport.h"
 #include "mcrouter/lib/network/UniqueIntrusiveList.h"
 #include "mcrouter/options.h"
 
@@ -63,7 +65,7 @@ class ProxyRequestContextTyped;
 class ShardSplitter;
 
 struct ProxyMessage {
-  enum class Type { REQUEST, OLD_CONFIG, SHUTDOWN };
+  enum class Type { REQUEST, OLD_CONFIG, REPLACE_AP, SHUTDOWN };
 
   Type type{Type::REQUEST};
   void* data{nullptr};
@@ -73,6 +75,35 @@ struct ProxyMessage {
   ProxyMessage(Type t, void* d) noexcept : type(t), data(d) {}
 };
 
+// struct used for replace message
+//
+struct replace_ap_t {
+  explicit replace_ap_t(
+      const AccessPoint& oldAp,
+      std::shared_ptr<const AccessPoint> newAp)
+      : oldAccessPoint(oldAp), newAccessPoint(std::move(newAp)) {}
+
+  // oldAccessPoint contains IP+port+other AP related info of the host that just
+  //                got removed
+  AccessPoint oldAccessPoint;
+  // newAccessPoint contains IP+port+other AP related info of the host that just
+  //                got added
+  std::shared_ptr<const AccessPoint> newAccessPoint;
+  // replacedAccessPoint
+  //                is the RETURN value from replace command which contains the
+  //                actual access point corresponding to the removed host
+  //                which is in the internal data structures like
+  //                ProxyDestination, ProxyDestinationMap, AccessPoints set in
+  //                ProxyConfig.
+  //                Since this is the actual AP present in the internal data
+  //                structures, we need to find this using the oldAccessPoint
+  //                and querying ProxyDestinationMap
+  std::shared_ptr<const AccessPoint> replacedAccessPoint;
+
+  // baton to wait on for the completion of the message processing
+  folly::fibers::Baton baton;
+};
+
 template <class RouterInfo>
 class Proxy : public ProxyBase {
  public:
@@ -131,6 +162,25 @@ class Proxy : public ProxyBase {
     return requestStats_.dump(filterZeroes);
   }
 
+  void processReplaceMessage(replace_ap_t* rep);
+
+  // replaceAP
+  // @oldAP    temporary AP constructed from IP address of the host
+  //           that got replaced
+  // @newAP    new AP that is used to replace oldAP in the ProxyDestination and
+  //           ProxyDestinationMap
+  //
+  // Returns
+  // replacedAP    AP in ProxyDestinationMap corresponding to the host that got
+  //           replaced, this is the real AP to be replaced. oldAP is used
+  //           to find this real AP
+  // Replace message is sent to proxy thread to find the AP to be replaced
+  // using 'oldAP' and then the AP is replaced by newAP and the replaced AP
+  // is returned to caller
+  std::shared_ptr<const AccessPoint> replaceAP(
+      const AccessPoint& oldAP,
+      std::shared_ptr<const AccessPoint> newAP);
+
   void advanceRequestStatsBin() override {
     requestStats().advanceBin();
   }

mcrouter/ProxyConfig.h

@@ -79,9 +79,27 @@ class ProxyConfig {
     return it->second.first;
   }
 
-  const folly::StringKeyedUnorderedMap<
-      std::vector<std::shared_ptr<const AccessPoint>>>&
-  getAccessPoints() const {
+  bool updateAccessPoints(
+      const std::string& pool,
+      std::shared_ptr<const AccessPoint>& oldAccessPoint,
+      std::shared_ptr<const AccessPoint>& newAccessPoint) {
+    auto it = accessPoints_.find(pool);
+    if (it != accessPoints_.end()) {
+      auto apIt = it->second.find(oldAccessPoint);
+      if (apIt != it->second.end()) {
+        it->second.erase(apIt);
+      } else {
+        return false;
+      }
+      it->second.insert(newAccessPoint);
+      return true;
+    }
+    return false;
+  }
+
+  folly::StringKeyedUnorderedMap<
+      std::unordered_set<std::shared_ptr<const AccessPoint>>>&
+  getAccessPoints() {
     return accessPoints_;
   }
 
@@ -92,7 +110,7 @@ class ProxyConfig {
   // config (after all RouteHandles) because its keys are being referenced
   // by object in the Config.
   folly::StringKeyedUnorderedMap<
-      std::vector<std::shared_ptr<const AccessPoint>>>
+      std::unordered_set<std::shared_ptr<const AccessPoint>>>
       accessPoints_;
 
   // pool source name -> (allow_partial_reconfig, [(pool_config,[pool_names])])

mcrouter/ProxyDestination.h

@@ -86,6 +86,13 @@ class ProxyDestination : public ProxyDestinationBase {
     return selfPtr_;
   }
 
+  std::shared_ptr<const AccessPoint> replaceAP(
+      std::shared_ptr<const AccessPoint> newAccessPoint) {
+    auto ret = ProxyDestinationBase::replaceAP(newAccessPoint);
+    closeGracefully();
+    return ret;
+  }
+
  private:
   std::unique_ptr<Transport, typename Transport::Destructor> transport_;
   // Ensure proxy thread doesn't reset the Transport

mcrouter/ProxyDestinationBase.cpp

@@ -114,7 +114,7 @@ bool ProxyDestinationBase::maySend(carbon::Result& tkoReason) const {
 void ProxyDestinationBase::onTkoEvent(TkoLogEvent event, carbon::Result result)
     const {
   auto logUtil = [this, result](folly::StringPiece eventStr) {
-    VLOG(2) << accessPoint_->toHostPortString() << " " << eventStr
+    VLOG(2) << accessPoint_.load()->toHostPortString() << " " << eventStr
             << ". Total hard TKOs: " << tracker_->globalTkos().hardTkos
             << "; soft TKOs: " << tracker_->globalTkos().softTkos
             << ". Reply: " << carbon::resultToString(result);
@@ -135,7 +135,7 @@ void ProxyDestinationBase::onTkoEvent(TkoLogEvent event, carbon::Result result)
       break;
   }
 
-  TkoLog tkoLog(*accessPoint_, tracker_->globalTkos());
+  TkoLog tkoLog(*accessPoint_.load(), tracker_->globalTkos());
   tkoLog.event = event;
   tkoLog.isHardTko = tracker_->isHardTko();
   tkoLog.isSoftTko = tracker_->isSoftTko();

mcrouter/ProxyDestinationBase.h

@@ -12,6 +12,7 @@
 #include <memory>
 
 #include <folly/IntrusiveList.h>
+#include <folly/concurrency/AtomicSharedPtr.h>
 
 #include "mcrouter/ExponentialSmoothData.h"
 #include "mcrouter/TkoLog.h"
@@ -97,8 +98,8 @@ class ProxyDestinationBase {
   /**
    * Destination host that this client talks to.
    */
-  const std::shared_ptr<const AccessPoint>& accessPoint() const {
-    return accessPoint_;
+  const std::shared_ptr<const AccessPoint> accessPoint() const {
+    return accessPoint_.load();
   }
 
   /**
@@ -174,12 +175,17 @@ class ProxyDestinationBase {
     return probeInflight_;
   }
 
+  std::shared_ptr<const AccessPoint> replaceAP(
+      std::shared_ptr<const AccessPoint> newAccessPoint) {
+    return accessPoint_.exchange(newAccessPoint);
+  }
+
  private:
   ProxyBase& proxy_;
   std::shared_ptr<TkoTracker> tracker_;
 
   // Destination host information
-  const std::shared_ptr<const AccessPoint> accessPoint_;
+  folly::atomic_shared_ptr<const AccessPoint> accessPoint_;
   std::chrono::milliseconds shortestConnectTimeout_{0};
   std::chrono::milliseconds shortestWriteTimeout_{0};
   const uint32_t qosClass_{0};

mcrouter/ProxyDestinationKey.cpp

@@ -22,6 +22,9 @@ ProxyDestinationKey::ProxyDestinationKey(
     std::chrono::milliseconds timeout_)
     : accessPoint(ap), timeout(timeout_) {}
 
+ProxyDestinationKey::ProxyDestinationKey(const AccessPoint& ap)
+    : ProxyDestinationKey(ap, std::chrono::milliseconds{0}) {}
+
 std::string ProxyDestinationKey::str() const {
   if (accessPoint.getProtocol() == mc_ascii_protocol) {
     // we cannot send requests with different timeouts for ASCII, since

mcrouter/ProxyDestinationKey.h

@@ -25,6 +25,7 @@ struct ProxyDestinationKey {
 
   explicit ProxyDestinationKey(const ProxyDestinationBase& dst);
   ProxyDestinationKey(const AccessPoint& ap, std::chrono::milliseconds timeout);
+  explicit ProxyDestinationKey(const AccessPoint& ap);
 
   std::string str() const;
   size_t hash() const;

mcrouter/ProxyDestinationMap-inl.h

@@ -20,6 +20,33 @@ namespace facebook {
 namespace memcache {
 namespace mcrouter {
 
+template <class Transport>
+std::shared_ptr<const AccessPoint> ProxyDestinationMap::replace(
+    const AccessPoint& tmpAccessPoint,
+    std::shared_ptr<const AccessPoint> newAccessPoint,
+    const std::chrono::milliseconds timeout) {
+  // Find the destination with old access point using
+  // temp access point. If it is found, the old access point is replaced
+  // with new access point
+  std::lock_guard<std::mutex> lck(destinationsLock_);
+  auto it = destinations_.find(ProxyDestinationKey(tmpAccessPoint, timeout));
+  if (it != destinations_.end()) {
+    auto destination = std::dynamic_pointer_cast<ProxyDestination<Transport>>(
+        (*it)->selfPtr().lock());
+    // Erase the destination before the access point is replaced because
+    // the internal logic to identify the destination inside destinations_
+    // uses access point info as Key
+    destinations_.erase(destination.get());
+    auto oldAccessPoint = destination->replaceAP(newAccessPoint);
+    // After the access point is replaced, insert the destination back into
+    // destinations with new access point used as Key
+    destinations_.emplace(destination.get());
+    // force the destination in TKO state
+    destination->handleTko(carbon::Result::CONNECT_ERROR, false);
+    return oldAccessPoint;
+  }
+  return nullptr;
+}
 template <class Transport>
 std::shared_ptr<ProxyDestination<Transport>> ProxyDestinationMap::emplace(
     std::shared_ptr<AccessPoint> ap,

mcrouter/ProxyDestinationMap.h

@@ -68,6 +68,12 @@ class ProxyDestinationMap {
       uint32_t qosPath,
       const std::shared_ptr<PoolTkoTracker>& poolTkoTracker);
 
+  template <class Transport>
+  std::shared_ptr<const AccessPoint> replace(
+      const AccessPoint& tmpOldAccessPoint,
+      std::shared_ptr<const AccessPoint> newAccessPoint,
+      std::chrono::milliseconds timeout);
+
   std::shared_ptr<PoolTkoTracker> createPoolTkoTracker(
       std::string poolName,
       uint32_t numEnterSoftTkos,