text: /**
   * Copyright - See the COPYRIGHT that is included with this distribution.
   * pvxs is distributed subject to a Software License Agreement found
   * in file LICENSE that is included with this distribution.
   */
  
  #include <algorithm>
  #include <set>
  #include <tuple>
  
  #include <osiSock.h>
  #include <dbDefs.h>
  #include <epicsThread.h>
  #include <epicsGuard.h>
  
  #include <pvxs/log.h>
  #include <clientimpl.h>
  
  DEFINE_LOGGER(setup, "pvxs.client.setup");
  DEFINE_LOGGER(io, "pvxs.client.io");
  DEFINE_LOGGER(duppv, "pvxs.client.dup");
  
  typedef epicsGuard<epicsMutex> Guard;
  typedef epicsGuardRelease<epicsMutex> UnGuard;
  
  namespace pvxs {
  namespace client {
  
  constexpr timeval bucketInterval{1,0};
  constexpr size_t nBuckets = 30u;
  
  constexpr size_t maxSearchPayload = 0x4000;
  
  constexpr timeval beaconCleanInterval{2*180, 0};
  
  Disconnect::Disconnect()
      :std::runtime_error("Disconnected")
      ,time(epicsTime::getCurrent())
  {}
  
  Disconnect::~Disconnect() {}
  
  RemoteError::RemoteError(const std::string& msg)
      :std::runtime_error(msg)
  {}
  
  RemoteError::~RemoteError() {}
  
  Finished::~Finished() {}
  
  Connected::Connected(const std::string& peerName)
      :std::runtime_error("Connected")
      ,peerName(peerName)
      ,time(epicsTime::getCurrent())
  {}
  
  Connected::~Connected() {}
  
  Interrupted::Interrupted()
      :std::runtime_error ("Interrupted")
  {}
  Interrupted::~Interrupted() {}
  
  Timeout::Timeout()
      :std::runtime_error ("Interrupted")
  {}
  Timeout::~Timeout() {}
  
  Channel::Channel(const std::shared_ptr<Context::Pvt>& context, const std::string& name, uint32_t cid)
      :context(context)
      ,name(name)
      ,cid(cid)
  {}
  
  Channel::~Channel()
  {
      context->chanByCID.erase(cid);
      context->chanByName.erase(name);
      // searchBuckets cleaned in tickSearch()
      if((state==Creating || state==Active) && conn && conn->bev) {
          {
              (void)evbuffer_drain(conn->txBody.get(), evbuffer_get_length(conn->txBody.get()));
  
              EvOutBuf R(hostBE, conn->txBody.get());
  
              to_wire(R, sid);
              to_wire(R, cid);
          }
          conn->enqueueTxBody(CMD_DESTROY_CHANNEL);
      }
  }
  
  void Channel::createOperations()
  {
      if(state!=Channel::Active)
          return;
  
      auto todo = std::move(pending);
  
      for(auto& wop : todo) {
          auto op = wop.lock();
          if(!op)
              continue;
  
          uint32_t ioid;
          do {
              ioid = conn->nextIOID++;
          } while(conn->opByIOID.find(ioid)!=conn->opByIOID.end());
  
          //conn->opByIOID.insert(std::make_pair(ioid, RequestInfo(sid, ioid, op)));
          auto pair = conn->opByIOID.emplace(std::piecewise_construct,
                                             std::forward_as_tuple(ioid),
                                             std::forward_as_tuple(sid, ioid, op));
          opByIOID[ioid] = &pair.first->second;
  
          op->ioid = ioid;
  
          op->createOp();
      }
  }
  
  void Channel::disconnect(const std::shared_ptr<Channel>& self)
  {
      self->state = Channel::Searching;
      self->sid = 0xdeadbeef; // spoil
      context->searchBuckets[context->currentBucket].push_back(self);
  
      log_debug_printf(io, "Server %s detach channel '%s' to re-search\n",
                       conn ? conn->peerName.c_str() : "<disconnected>",
                       self->name.c_str());
  
  }
  
  Value ResultWaiter::wait(double timeout)
  {
      Guard G(lock);
      while(outcome==Busy) {
          UnGuard U(G);
          if(!notify.wait(timeout))
              throw Timeout();
      }
      if(outcome==Done)
          return result();
      else
          throw Interrupted();
  }
  
  void ResultWaiter::complete(Result&& result, bool interrupt)
  {
      bool wakeup;
      {
          Guard G(lock);
          wakeup = outcome==Busy;
          if(wakeup) {
              this->result = std::move(result);
              outcome = interrupt ? Abort : Done;
          }
      }
      if(wakeup)
          notify.signal();
  }
  
  OperationBase::OperationBase(operation_t op, const std::shared_ptr<Channel>& chan)
      :Operation(op)
      ,chan(chan)
  {}
  
  OperationBase::~OperationBase() {}
  
  Value OperationBase::wait(double timeout)
  {
      if(!waiter)
          throw std::logic_error("Operation has custom .result() callback");
      return waiter->wait(timeout);
  }
  
  void OperationBase::interrupt()
  {
      if(waiter)
          waiter->complete(Result(), true);
  }
  
  RequestInfo::RequestInfo(uint32_t sid, uint32_t ioid, std::shared_ptr<OperationBase>& handle)
      :sid(sid)
      ,ioid(ioid)
      ,op(handle->op)
      ,handle(handle)
  {}
  
  std::shared_ptr<Channel> Channel::build(const std::shared_ptr<Context::Pvt>& context, const std::string& name)
  {
  
      std::shared_ptr<Channel> chan;
  
      auto it = context->chanByName.find(name);
      if(it!=context->chanByName.end()) {
          chan = it->second.lock();
      }
  
      if(!chan) {
          while(context->chanByCID.find(context->nextCID)!=context->chanByCID.end())
              context->nextCID++;
  
          chan = std::make_shared<Channel>(context, name, context->nextCID);
          context->chanByCID[chan->cid] = chan;
          context->chanByName[chan->name] = chan;
  
          context->searchBuckets[context->currentBucket].push_back(chan);
      }
  
      return chan;
  }
  
  Operation::~Operation() {}
  
  Subscription::~Subscription() {}
  
  Context::Context(const Config& conf)
  {
      /* Here be dragons.
       *
       * We keep two different ref. counters.
       * - "externel" counter which keeps a server running.
       * - "internal" which only keeps server storage from being destroyed.
       *
       * External refs are held as Server::pvt.  Internal refs are
       * held by various in-progress operations (OpBase sub-classes)
       * Which need to safely access server storage, but should not
       * prevent a server from stopping.
       */
      auto internal(std::make_shared<Pvt>(conf));
      internal->internal_self = internal;
      cnt_ClientPvtLive.fetch_add(1u);
  
      // external
      pvt.reset(internal.get(), [internal](Pvt*) mutable {
          auto temp(std::move(internal));
          try {
              temp->close();
          }catch(std::exception& e){
              // called through ~shared_ptr and can't propagate exceptions.
              // log and continue...
              log_exc_printf(setup, "Error while closing Context (%s) : %s\n",
                             typeid(e).name(), e.what());
          }
          cnt_ClientPvtLive.fetch_sub(1u);
      });
      // we don't keep a weak_ptr to the external reference.
      // Caller is entirely responsible for keeping this server running
  }
  
  Context::~Context() {}
  
  const Config& Context::config() const
  {
      if(!pvt)
          throw std::logic_error("NULL Context");
  
      return pvt->effective;
  }
  
  void Context::hurryUp()
  {
      if(!pvt)
          throw std::logic_error("NULL Context");
  
      pvt->manager.loop().call([this](){
          pvt->poke();
      });
  }
  
  static
  Value buildCAMethod()
  {
      using namespace pvxs::members;
  
      return TypeDef(TypeCode::Struct, {
                         String("user"),
                         String("host"),
                     }).create();
  }
  
  Context::Pvt::Pvt(const Config& conf)
      :effective(conf)
      ,caMethod(buildCAMethod())
      ,searchTx(AF_INET, SOCK_DGRAM, 0)
      ,tcp_loop("PVXCTCP", epicsThreadPriorityCAServerLow)
      ,searchRx(event_new(tcp_loop.base, searchTx.sock, EV_READ|EV_PERSIST, &Pvt::onSearchS, this))
      ,searchTimer(event_new(tcp_loop.base, -1, EV_TIMEOUT, &Pvt::tickSearchS, this))
      ,manager(UDPManager::instance())
      ,beaconCleaner(event_new(manager.loop().base, -1, EV_TIMEOUT|EV_PERSIST, &Pvt::tickBeaconCleanS, this))
  {
      effective.expand();
  
      searchBuckets.resize(nBuckets);
  
      std::set<std::string> bcasts;
      {
          ELLLIST list = ELLLIST_INIT;
          osiSockAddr any{};
  
          osiSockDiscoverBroadcastAddresses(&list, searchTx.sock, &any);
  
          while(ELLNODE *cur = ellGet(&list)) {
              osiSockAddrNode *node = CONTAINER(cur, osiSockAddrNode, node);
  
              SockAddr addr(&node->addr.sa, sizeof(node->addr));
              addr.setPort(0u);
              bcasts.insert(addr.tostring());
  
              free(node);
          }
      }
  
      {
          osiSockAddr any{};
          any.ia.sin_family = AF_INET;
          if(bind(searchTx.sock, &any.sa, sizeof(any.ia)))
              throw std::runtime_error("Unable to bind random UDP port");
  
          socklen_t alen = sizeof(any);
          if(getsockname(searchTx.sock, &any.sa, &alen))
              throw std::runtime_error("Unable to readback random UDP port");
  
          searchRxPort = ntohs(any.ia.sin_port);
  
          log_debug_printf(setup, "Using UDP Rx port %u\n", searchRxPort);
      }
  
      {
          int val = 1;
          if(setsockopt(searchTx.sock, SOL_SOCKET, SO_BROADCAST, (char *)&val, sizeof(val)))
              log_err_printf(setup, "Unable to setup beacon sender SO_BROADCAST: %d\n", SOCKERRNO);
      }
  
      for(auto& addr : effective.addressList) {
          auto isbcast = bcasts.find(addr)!=bcasts.end();
          SockAddr saddr(AF_INET);
          try {
              saddr.setAddress(addr.c_str(), effective.udp_port);
          }catch(std::runtime_error& e) {
              log_err_printf(setup, "%s  Ignoring...\n", e.what());
          }
          auto top = ntohl(saddr->in.sin_addr.s_addr)>>24u;
          auto isucast = !isbcast && top<239 && top>224;
  
          searchDest.emplace_back(saddr, isucast);
      }
  
      for(auto& iface : effective.interfaces) {
          SockAddr addr(AF_INET, iface.c_str(), effective.udp_port);
          log_info_printf(io, "Listening for beacons on %s\n", addr.tostring().c_str());
          beaconRx.push_back(manager.onBeacon(addr, [this](const UDPManager::Beacon& msg) {
              onBeacon(msg);
          }));
      }
  
      for(auto& listener : beaconRx) {
          listener->start();
      }
  
      if(event_add(searchTimer.get(), &bucketInterval))
          log_err_printf(setup, "Error enabling search timer\n%s", "");
      if(event_add(searchRx.get(), nullptr))
          log_err_printf(setup, "Error enabling search RX\n%s", "");
      if(event_add(beaconCleaner.get(), &beaconCleanInterval))
          log_err_printf(setup, "Error enabling beacon clean timer on\n%s", "");
  }
  
  Context::Pvt::~Pvt() {}
  
  void Context::Pvt::close()
  {
      // terminate all active connections
      tcp_loop.call([this]() {
          (void)event_del(searchTimer.get());
          (void)event_del(searchRx.get());
          (void)event_del(beaconCleaner.get());
  
          decltype (connByAddr) conns(std::move(connByAddr));
  
          for(auto& pair : conns) {
              auto conn = pair.second.lock();
              if(!conn)
                  continue;
  
              conn->cleanup();
          }
      });
  
      tcp_loop.join();
  
      // ensure any in-progress callbacks have completed
      manager.sync();
  }
  
  void Context::Pvt::poke()
  {
      epicsTimeStamp now{};
  
      double age = -1.0;
      if(epicsTimeGetCurrent(&now) || (age=epicsTimeDiffInSeconds(&now, &lastPoke))<30.0) {
          log_debug_printf(setup, "Ignoring hurryUp() age=%.1f sec\n", age);
          return;
      }
      lastPoke = now;
  
      log_debug_printf(setup, "hurryUp()%s\n", "");
  
      timeval immediate{0,0};
      if(event_add(searchTimer.get(), &immediate))
          throw std::runtime_error("Unable to schedule searchTimer");
  }
  
  void Context::Pvt::onBeacon(const UDPManager::Beacon& msg)
  {
      const auto& guid = msg.guid;
  
      epicsTimeStamp now;
      epicsTimeGetCurrent(&now);
  
      auto it = beaconSenders.find(msg.src);
      if(it!=beaconSenders.end() && msg.guid==it->second.guid) {
          it->second.lastRx = now;
          return;
      }
  
      beaconSenders.emplace(msg.src, BTrack{msg.guid, now});
  
      log_debug_printf(io, "%s New server %02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x %s\n",
                 msg.src.tostring().c_str(),
                 guid[0], guid[1], guid[2], guid[3], guid[4], guid[5], guid[6], guid[7], guid[8], guid[9], guid[10], guid[11],
                 msg.server.tostring().c_str());
  
      poke();
  }
  
  bool Context::Pvt::onSearch()
  {
      searchMsg.resize(0x10000);
      SockAddr src;
  
      osiSocklen_t alen = src.size();
      const int nrx = recvfrom(searchTx.sock, (char*)&searchMsg[0], searchMsg.size()-1, 0, &src->sa, &alen);
  
      if(nrx<0) {
          int err = evutil_socket_geterror(searchTx.sock);
          if(err==SOCK_EWOULDBLOCK || err==EAGAIN || err==SOCK_EINTR) {
              // nothing to do here
          } else {
              log_warn_printf(io, "UDP search RX Error on : %s\n",
                         evutil_socket_error_to_string(err));
          }
          return false; // wait for more I/O
  
      } else if(nrx<8) {
          // maybe a zero (body) length packet?
          // maybe an OS error?
  
          log_info_printf(io, "UDP ignore runt%s\n", "");
          return true;
  
      } else if(searchMsg[0]!=0xca || searchMsg[1]==0 || (searchMsg[2]&(pva_flags::Control|pva_flags::SegMask))) {
          // minimum header size is 8 bytes
          // ID byte must by 0xCA (because PVA has some paternal envy)
          // ignore incompatible version 0
          // UDP packets can't contain control messages, or use segmentation
  
          log_info_printf(io, "UDP ignore header%u %02x%02x%02x%02x\n",
                     unsigned(nrx), searchMsg[0], searchMsg[1], searchMsg[2], searchMsg[3]);
          return true;
      }
  
      log_hex_printf(io, Level::Debug, &searchMsg[0], nrx, "UDP search Rx %d from %s\n", nrx, src.tostring().c_str());
  
      bool be = searchMsg[2]&pva_flags::MSB;
  
      FixedBuf M(be, searchMsg.data(), nrx);
  
      const uint8_t cmd = M[3];
      M.skip(4);
  
      uint32_t len=0;
      from_wire(M, len);
  
      if(len > M.size() && M.good()) {
          log_info_printf(io, "UDP ignore header%u %02x%02x%02x%02x\n",
                     unsigned(M.size()), M[0], M[1], M[2], M[3]);
          return true;
      }
  
      if(cmd==CMD_SEARCH_RESPONSE) {
          std::array<uint8_t, 12> guid;
          SockAddr serv;
          uint16_t port = 0;
          uint8_t found = 0u;
  
          _from_wire<12>(M, &guid[0], false);
          // searchSequenceID
          // we don't use this and instead rely on ID for individual PVs
          M.skip(4u);
  
          from_wire(M, serv);
          if(serv.isAny())
              serv = src;
          from_wire(M, port);
          serv.setPort(port);
  
          if(M.size()<4u || M[0]!=3u || M[1]!='t' || M[2]!='c' || M[3]!='p')
              return true;
          M.skip(4u);
  
          from_wire(M, found);
          if(!found)
              return true;
  
          uint16_t nSearch = 0u;
          from_wire(M, nSearch);
  
          for(auto n : range(nSearch)) {
              (void)n;
  
              uint32_t id=0u;
              from_wire(M, id);
              if(!M.good())
                  break;
  
              std::shared_ptr<Channel> chan;
              {
                  auto it = chanByCID.find(id);
                  if(it==chanByCID.end())
                      continue;
  
                  chan = it->second.lock();
                  if(!chan)
                      continue;
              }
  
              log_debug_printf(io, "Search reply for %s\n", chan->name.c_str());
  
              if(chan->state==Channel::Searching) {
                  chan->guid = guid;
                  chan->replyAddr = serv;
  
                  auto it = connByAddr.find(serv);
                  if(it==connByAddr.end() || !(chan->conn = it->second.lock())) {
                      connByAddr[serv] = chan->conn = std::make_shared<Connection>(internal_self.lock(), serv);
                  }
  
                  chan->conn->pending.push_back(chan);
                  chan->state = Channel::Connecting;
  
                  chan->conn->createChannels();
  
              } else if(chan->guid!=guid) {
                  log_err_printf(duppv, "Duplicate PV name %s from %s and %s\n",
                                 chan->name.c_str(),
                                 chan->replyAddr.tostring().c_str(),
                                 serv.tostring().c_str());
              }
          }
  
      } else {
          M.fault();
      }
  
      if(!M.good()) {
          log_hex_printf(io, Level::Err, &searchMsg[0], nrx, "Invalid search reply %d from %s\n", nrx, src.tostring().c_str());
      }
  
      return true;
  }
  
  void Context::Pvt::onSearchS(evutil_socket_t fd, short evt, void *raw)
  {
      try {
          log_debug_printf(io, "UDP search Rx event %x\n", evt);
          if(!(evt&EV_READ))
              return;
  
          // handle up to 4 packets before going back to the reactor
          for(unsigned i=0; i<4 && static_cast<Pvt*>(raw)->onSearch(); i++) {}
  
      }catch(std::exception& e){
          log_exc_printf(io, "Unhandled error in search Rx callback: %s\n", e.what());
      }
  }
  
  void Context::Pvt::tickSearch()
  {
      auto idx = currentBucket;
      currentBucket = (currentBucket+1u)%searchBuckets.size();
  
      log_debug_printf(io, "Search tick %zu\n", idx);
  
      decltype (searchBuckets)::value_type bucket;
      searchBuckets[idx].swap(bucket);
  
      while(!bucket.empty()) {
          searchMsg.resize(0x10000);
          FixedBuf M(true, searchMsg.data(), searchMsg.size());
          M.skip(8); // fill in header after body length known
  
          // searchSequenceID
          // we don't use this and instead rely on IDs for individual PVs
          to_wire(M, uint32_t(0x66696e64));
  
          // flags and reserved.
          // initially flags[7] is cleared (bcast)
          auto pflags = M.save();
          to_wire(M, uint32_t(0u));
  
          // IN6ADDR_ANY_INIT
          to_wire(M, uint32_t(0u));
          to_wire(M, uint32_t(0u));
          to_wire(M, uint32_t(0u));
          to_wire(M, uint32_t(0u));
  
          to_wire(M, uint16_t(searchRxPort));
  
          to_wire(M, uint8_t(1u));
          to_wire(M, "tcp");
  
          // placeholder for channel count;
          auto pcount = M.save();
          uint16_t count = 0u;
          M.skip(2u);
  
          bool payload = false;
          while(!bucket.empty()) {
              auto chan = bucket.front().lock();
              if(!chan || chan->state!=Channel::Searching) {
                  bucket.pop_front();
                  continue;
              }
  
              if(searchMsg.size()<=maxSearchPayload-(5+chan->name.size()))
                  break;
  
              to_wire(M, uint32_t(chan->cid));
              to_wire(M, chan->name);
              count++;
  
              auto ninc = chan->nSearch = std::min(searchBuckets.size(), chan->nSearch+1u);
              auto next = (idx + ninc)%searchBuckets.size();
              auto nextnext = (next + 1u)%searchBuckets.size();
  
              // try to smooth out UDP bcast load by waiting one extra tick
              {
                  auto nextN = searchBuckets[next].size();
                  auto nextnextN = searchBuckets[nextnext].size();
  
                  if(nextN > nextnextN && (nextN-nextnextN > 100u))
                      next = nextnext;
              }
  
              auto& nextBucket = searchBuckets[next];
  
              nextBucket.splice(nextBucket.end(),
                                bucket,
                                bucket.begin());
              payload = true;
          }
  
          if(!payload)
              break;
  
          {
              FixedBuf C(true, pcount, 2u);
              to_wire(C, count);
          }
          auto consumed = M.save() - searchMsg.data();
          {
              FixedBuf H(true, searchMsg.data(), 8);
              to_wire(H, Header{CMD_SEARCH, pva_flags::Server, uint32_t(consumed-8u)});
          }
          for(auto& pair : searchDest) {
              *pflags = pair.second ? 0x80 : 0x00;
  
              int ntx = sendto(searchTx.sock, (char*)searchMsg.data(), consumed, 0, &pair.first->sa, pair.first.size());
  
              if(ntx<0) {
                  int err = evutil_socket_geterror(searchTx.sock);
                  auto lvl = Level::Warn;
                  if(err==EINTR || err==EPERM)
                      lvl = Level::Debug;
                  log_printf(io, lvl, "Search tx error (%d) %s\n",
                             err, evutil_socket_error_to_string(err));
  
              } else if(unsigned(ntx)<consumed) {
                  log_warn_printf(io, "Search truncated %u < %u",
                             unsigned(ntx), unsigned(consumed));
  
              } else {
                  log_debug_printf(io, "Search to %s %s\n", pair.first.tostring().c_str(),
                                   pair.second ? "ucast" : "bcast");
              }
          }
      }
  
      if(event_add(searchTimer.get(), &bucketInterval))
          log_err_printf(setup, "Error re-enabling search timer on\n%s", "");
  }
  
  void Context::Pvt::tickSearchS(evutil_socket_t fd, short evt, void *raw)
  {
      try {
          static_cast<Pvt*>(raw)->tickSearch();
      }catch(std::exception& e){
          log_exc_printf(io, "Unhandled error in search timer callback: %s\n", e.what());
      }
  }
  
  void Context::Pvt::tickBeaconClean()
  {
      epicsTimeStamp now;
      epicsTimeGetCurrent(&now);
  
      auto it = beaconSenders.begin();
      while(it!=beaconSenders.end()) {
          auto cur = it++;
  
          double age = epicsTimeDiffInSeconds(&now, &cur->second.lastRx);
  
          if(age < -15.0 || age > 2.1*180.0) {
              auto& guid = cur->second.guid;
              log_debug_printf(io, "Lost server %02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x %s\n",
                         guid[0], guid[1], guid[2], guid[3], guid[4], guid[5], guid[6], guid[7], guid[8], guid[9], guid[10], guid[11],
                         cur->first.tostring().c_str());
  
              beaconSenders.erase(cur);
          }
      }
  }
  
  void Context::Pvt::tickBeaconCleanS(evutil_socket_t fd, short evt, void *raw)
  {
      try {
          static_cast<Pvt*>(raw)->tickBeaconClean();
      }catch(std::exception& e){
          log_exc_printf(io, "Unhandled error in beacon cleaner timer callback: %s\n", e.what());
      }
  }
  
  } // namespace client
  
  } // namespace pvxs
  