text: #include "stdafx.h"
  #include "udp_http.h"
  #include "udp_client_server.h"
  #include "udp_socket.h"
  #include "cpu_affinity.h"
  
  #include <util/system/hp_timer.h>
  #include <util/thread/lfqueue.h>
  #include <util/system/thread.h>
  #include <util/system/spinlock.h>
  #if !defined(_win_)
  #include <signal.h>
  #include <pthread.h>
  #endif
  #include "block_chain.h"
  #include <util/system/shmat.h>
  
  namespace NNetliba {
      const float HTTP_TIMEOUT = 15.0f;
      const int MIN_SHARED_MEM_PACKET = 1000;
  
      static volatile bool PanicAttack;
      static volatile NHPTimer::STime LastHeartbeat;
      static volatile double HeartbeatTimeout;
  
      static int GetPacketSize(TRequest* req) {
          if (req && req->Data.Get())
              return req->Data->GetSize();
          return 0;
      }
  
      static bool IsLocalFast(const TUdpAddress& addr) {
          if (addr.IsIPv4()) {
              return IsLocalIPv4(addr.GetIPv4());
          } else {
              return IsLocalIPv6(addr.Network, addr.Interface);
          }
      }
  
      bool IsLocal(const TUdpAddress& addr) {
          InitLocalIPList();
          return IsLocalFast(addr);
      }
  
      TUdpHttpRequest::~TUdpHttpRequest() {
      }
  
      TUdpHttpResponse::~TUdpHttpResponse() {
      }
  
      class TRequesterUserQueueSizes: public TThrRefBase {
      public:
          TAtomic ReqCount, RespCount;
          TAtomic ReqQueueSize, RespQueueSize;
  
          TRequesterUserQueueSizes()
              : ReqCount(0)
              , RespCount(0)
              , ReqQueueSize(0)
              , RespQueueSize(0)
          {
          }
      };
  
      template <class T>
      void EraseList(TLockFreeQueue<T*>* data) {
          T* ptr = nullptr;
          while (data->Dequeue(&ptr)) {
              delete ptr;
          }
      }
  
      class TRequesterUserQueues: public TThrRefBase {
          TIntrusivePtr<TRequesterUserQueueSizes> QueueSizes;
          TLockFreeQueue<TUdpHttpRequest*> ReqList;
          TLockFreeQueue<TUdpHttpResponse*> ResponseList;
          TLockFreeStack<TGUID> CancelList, SendRequestAccList; // any order will do
          TMuxEvent AsyncEvent;
  
          void UpdateAsyncSignalState() {
              // not sure about this one. Idea is that AsyncEvent.Reset() is a memory barrier
              if (ReqList.IsEmpty() && ResponseList.IsEmpty() && CancelList.IsEmpty() && SendRequestAccList.IsEmpty()) {
                  AsyncEvent.Reset();
                  if (!ReqList.IsEmpty() || !ResponseList.IsEmpty() || !CancelList.IsEmpty() || !SendRequestAccList.IsEmpty())
                      AsyncEvent.Signal();
              }
          }
          ~TRequesterUserQueues() override {
              EraseList(&ReqList);
              EraseList(&ResponseList);
          }
  
      public:
          TRequesterUserQueues(TRequesterUserQueueSizes* queueSizes)
              : QueueSizes(queueSizes)
          {
          }
          TUdpHttpRequest* GetRequest();
          TUdpHttpResponse* GetResponse();
          bool GetRequestCancel(TGUID* req) {
              bool res = CancelList.Dequeue(req);
              UpdateAsyncSignalState();
              return res;
          }
          bool GetSendRequestAcc(TGUID* req) {
              bool res = SendRequestAccList.Dequeue(req);
              UpdateAsyncSignalState();
              return res;
          }
  
          void AddRequest(TUdpHttpRequest* res) {
              AtomicAdd(QueueSizes->ReqCount, 1);
              AtomicAdd(QueueSizes->ReqQueueSize, GetPacketSize(res->DataHolder.Get()));
              ReqList.Enqueue(res);
              AsyncEvent.Signal();
          }
          void AddResponse(TUdpHttpResponse* res) {
              AtomicAdd(QueueSizes->RespCount, 1);
              AtomicAdd(QueueSizes->RespQueueSize, GetPacketSize(res->DataHolder.Get()));
              ResponseList.Enqueue(res);
              AsyncEvent.Signal();
          }
          void AddCancel(const TGUID& req) {
              CancelList.Enqueue(req);
              AsyncEvent.Signal();
          }
          void AddSendRequestAcc(const TGUID& req) {
              SendRequestAccList.Enqueue(req);
              AsyncEvent.Signal();
          }
          TMuxEvent& GetAsyncEvent() {
              return AsyncEvent;
          }
          void AsyncSignal() {
              AsyncEvent.Signal();
          }
      };
  
      struct TOutRequestState {
          enum EState {
              S_SENDING,
              S_WAITING,
              S_WAITING_PING_SENDING,
              S_WAITING_PING_SENT,
              S_CANCEL_AFTER_SENDING
          };
          EState State;
          TUdpAddress Address;
          double TimePassed;
          int PingTransferId;
          TIntrusivePtr<TRequesterUserQueues> UserQueues;
  
          TOutRequestState()
              : State(S_SENDING)
              , TimePassed(0)
              , PingTransferId(-1)
          {
          }
      };
  
      struct TInRequestState {
          enum EState {
              S_WAITING,
              S_RESPONSE_SENDING,
              S_CANCELED,
          };
          EState State;
          TUdpAddress Address;
  
          TInRequestState()
              : State(S_WAITING)
          {
          }
          TInRequestState(const TUdpAddress& address)
              : State(S_WAITING)
              , Address(address)
          {
          }
      };
  
      enum EHttpPacket {
          PKT_REQUEST,
          PKT_PING,
          PKT_PING_RESPONSE,
          PKT_RESPONSE,
          PKT_GETDEBUGINFO,
          PKT_LOCAL_REQUEST,
          PKT_LOCAL_RESPONSE,
          PKT_CANCEL,
      };
  
      class TUdpHttp: public IRequester {
          enum EDir {
              DIR_OUT,
              DIR_IN
          };
          struct TTransferPurpose {
              EDir Dir;
              TGUID Guid;
              TTransferPurpose()
                  : Dir(DIR_OUT)
              {
              }
              TTransferPurpose(EDir dir, TGUID guid)
                  : Dir(dir)
                  , Guid(guid)
              {
              }
          };
  
          struct TSendRequest {
              TUdpAddress Addr;
              TAutoPtr<TRopeDataPacket> Data;
              TGUID ReqGuid;
              TIntrusivePtr<TWaitResponse> WR;
              TIntrusivePtr<TRequesterUserQueues> UserQueues;
              ui32 Crc32;
  
              TSendRequest()
                  : Crc32(0)
              {
              }
              TSendRequest(const TUdpAddress& addr, TAutoPtr<TRopeDataPacket>* data, const TGUID& reqguid, TWaitResponse* wr, TRequesterUserQueues* userQueues)
                  : Addr(addr)
                  , Data(*data)
                  , ReqGuid(reqguid)
                  , WR(wr)
                  , UserQueues(userQueues)
                  , Crc32(CalcChecksum(Data->GetChain()))
              {
              }
          };
          struct TSendResponse {
              TVector<char> Data;
              TGUID ReqGuid;
              ui32 DataCrc32;
              EPacketPriority Priority;
  
              TSendResponse()
                  : DataCrc32(0)
                  , Priority(PP_NORMAL)
              {
              }
              TSendResponse(const TGUID& reqguid, EPacketPriority prior, TVector<char>* data)
                  : ReqGuid(reqguid)
                  , DataCrc32(0)
                  , Priority(prior)
              {
                  if (data && !data->empty()) {
                      data->swap(Data);
                      DataCrc32 = TIncrementalChecksumCalcer::CalcBlockSum(&Data[0], Data.ysize());
                  }
              }
          };
          struct TCancelRequest {
              TGUID ReqGuid;
  
              TCancelRequest() = default;
              TCancelRequest(const TGUID& reqguid)
                  : ReqGuid(reqguid)
              {
              }
          };
          struct TBreakRequest {
              TGUID ReqGuid;
  
              TBreakRequest() = default;
              TBreakRequest(const TGUID& reqguid)
                  : ReqGuid(reqguid)
              {
              }
          };
  
          TThread myThread;
          bool KeepRunning, AbortTransactions;
          TSpinLock cs;
          TSystemEvent HasStarted;
  
          NHPTimer::STime PingsSendT;
  
          TIntrusivePtr<IUdpHost> Host;
          TIntrusivePtr<NNetlibaSocket::ISocket> Socket;
          typedef THashMap<TGUID, TOutRequestState, TGUIDHash> TOutRequestHash;
          typedef THashMap<TGUID, TInRequestState, TGUIDHash> TInRequestHash;
          TOutRequestHash OutRequests;
          TInRequestHash InRequests;
  
          typedef THashMap<int, TTransferPurpose> TTransferHash;
          TTransferHash TransferHash;
  
          typedef THashMap<TGUID, TIntrusivePtr<TWaitResponse>, TGUIDHash> TSyncRequests;
          TSyncRequests SyncRequests;
  
          // hold it here to not construct on every DoSends()
          typedef THashSet<TGUID, TGUIDHash> TAnticipateCancels;
          TAnticipateCancels AnticipateCancels;
  
          TLockFreeQueue<TSendRequest*> SendReqList;
          TLockFreeQueue<TSendResponse*> SendRespList;
          TLockFreeQueue<TCancelRequest> CancelReqList;
          TLockFreeQueue<TBreakRequest> BreakReqList;
  
          TIntrusivePtr<TRequesterUserQueueSizes> QueueSizes;
          TIntrusivePtr<TRequesterUserQueues> UserQueues;
  
          struct TStatsRequest: public TThrRefBase {
              enum EReq {
                  PENDING_SIZE,
                  DEBUG_INFO,
                  HAS_IN_REQUEST,
                  GET_PEER_ADDRESS,
                  GET_PEER_QUEUE_STATS,
              };
              EReq Req;
              TRequesterPendingDataStats PendingDataSize;
              TString DebugInfo;
              TGUID RequestId;
              TUdpAddress PeerAddress;
              TIntrusivePtr<IPeerQueueStats> QueueStats;
              bool RequestFound;
              TSystemEvent Complete;
  
              TStatsRequest(EReq req)
                  : Req(req)
                  , RequestFound(false)
              {
              }
          };
          TLockFreeQueue<TIntrusivePtr<TStatsRequest>> StatsReqList;
  
          bool ReportRequestCancel;
          bool ReportSendRequestAcc;
  
          void FinishRequest(TOutRequestHash::iterator i, TUdpHttpResponse::EResult ok, TAutoPtr<TRequest> data, const char* error = nullptr) {
              TOutRequestState& s = i->second;
              TUdpHttpResponse* res = new TUdpHttpResponse;
              res->DataHolder = data;
              res->ReqId = i->first;
              res->PeerAddress = s.Address;
              res->Ok = ok;
              if (ok == TUdpHttpResponse::FAILED)
                  res->Error = error ? error : "request failed";
              else if (ok == TUdpHttpResponse::CANCELED)
                  res->Error = error ? error : "request cancelled";
              TSyncRequests::iterator k = SyncRequests.find(res->ReqId);
              if (k != SyncRequests.end()) {
                  TIntrusivePtr<TWaitResponse>& wr = k->second;
                  wr->SetResponse(res);
                  SyncRequests.erase(k);
              } else {
                  s.UserQueues->AddResponse(res);
              }
  
              OutRequests.erase(i);
          }
          int SendWithHighPriority(const TUdpAddress& addr, TAutoPtr<TRopeDataPacket> data) {
              ui32 crc32 = CalcChecksum(data->GetChain());
              return Host->Send(addr, data.Release(), crc32, nullptr, PP_HIGH);
          }
          void ProcessIncomingPackets() {
              TVector<TGUID, TCustomAllocator<TGUID>> failedRequests;
              for (;;) {
                  TAutoPtr<TRequest> req = Host->GetRequest();
                  if (req.Get() == nullptr) {
                      if (!failedRequests.empty()) {
                          // we want to handle following sequence of events
                          // <- send ping
                          // -> send response over IB
                          // -> send ping response (no such request) over UDP
                          // Now if we are lucky enough we can get IB response waiting in the IB receive queue
                          // at the same time response sender will receive "send complete" from IB
                          // indeed, IB delivered message (but it was not parsed by ib_cs.cpp yet)
                          // so after receiving "send response complete" event resposne sender can legally response
                          // to pings with "no such request"
                          // but ping responses can be sent over UDP
                          // So we can run into situation with negative ping response in
                          // UDP receive queue and response waiting unprocessed in IB receive queue
                          // to check that there is no response in the IB queue we have to process IB queues
                          // so we call IBStep()
                          Host->IBStep();
                          req = Host->GetRequest();
                          if (req.Get() == nullptr) {
                              break;
                          }
                      } else {
                          break;
                      }
                  }
  
                  TBlockChainIterator reqData(req->Data->GetChain());
                  char pktType;
                  reqData.Read(&pktType, 1);
                  switch (pktType) {
                      case PKT_REQUEST:
                      case PKT_LOCAL_REQUEST: {
                          //printf("recv PKT_REQUEST or PKT_LOCAL_REQUEST\n");
                          TGUID reqId = req->Guid;
                          TInRequestHash::iterator z = InRequests.find(reqId);
                          if (z != InRequests.end()) {
                              // oops, this request already exists!
                              // might happen if request can be stored in single packet
                              // and this packet had source IP broken during transmission and managed to pass crc checks
                              // since we already reported wrong source address for this request to the user
                              // the best thing we can do is to stop the program to avoid further complications
                              // but we just report the accident to stderr
                              fprintf(stderr, "Jackpot, same request %s received twice from %s and earlier from %s\n",
                                      GetGuidAsString(reqId).c_str(), GetAddressAsString(z->second.Address).c_str(),
                                      GetAddressAsString(req->Address).c_str());
                          } else {
                              InRequests[reqId] = TInRequestState(req->Address);
  
                              //printf("InReq %s PKT_REQUEST recv ... -> S_WAITING\n", GetGuidAsString(reqId).c_str());
  
                              TUdpHttpRequest* res = new TUdpHttpRequest;
                              res->ReqId = reqId;
                              res->PeerAddress = req->Address;
                              res->DataHolder = req;
  
                              UserQueues->AddRequest(res);
                          }
                      } break;
                      case PKT_PING: {
                          //printf("recv PKT_PING\n");
                          TGUID guid;
                          reqData.Read(&guid, sizeof(guid));
                          bool ok = InRequests.find(guid) != InRequests.end();
                          TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
                          ms->Write((char)PKT_PING_RESPONSE);
                          ms->Write(guid);
                          ms->Write(ok);
                          SendWithHighPriority(req->Address, ms.Release());
                          //printf("InReq %s PKT_PING recv Sending PKT_PING_RESPONSE\n", GetGuidAsString(guid).c_str());
                          //printf("got PKT_PING, responding %d\n", (int)ok);
                      } break;
                      case PKT_PING_RESPONSE: {
                          //printf("recv PKT_PING_RESPONSE\n");
                          TGUID guid;
                          bool ok;
                          reqData.Read(&guid, sizeof(guid));
                          reqData.Read(&ok, sizeof(ok));
                          TOutRequestHash::iterator i = OutRequests.find(guid);
                          if (i == OutRequests.end()) {
                              ; //Y_ASSERT(0); // actually possible with some packet orders
                          } else {
                              if (!ok) {
                                  // can not delete request at this point
                                  // since we can receive failed ping and response at the same moment
                                  // consider sequence: client sends ping, server sends response
                                  // and replies false to ping as reply is sent
                                  // we can not receive failed ping_response earlier then response itself
                                  // but we can receive them simultaneously
                                  failedRequests.push_back(guid);
                                  //printf("OutReq %s PKT_PING_RESPONSE recv no such query -> failed\n", GetGuidAsString(guid).c_str());
                              } else {
                                  TOutRequestState& s = i->second;
                                  switch (s.State) {
                                      case TOutRequestState::S_WAITING_PING_SENDING: {
                                          Y_ASSERT(s.PingTransferId >= 0);
                                          TTransferHash::iterator k = TransferHash.find(s.PingTransferId);
                                          if (k != TransferHash.end())
                                              TransferHash.erase(k);
                                          else
                                              Y_ASSERT(0);
                                          s.PingTransferId = -1;
                                          s.TimePassed = 0;
                                          s.State = TOutRequestState::S_WAITING;
                                          //printf("OutReq %s PKT_PING_RESPONSE recv S_WAITING_PING_SENDING -> S_WAITING\n", GetGuidAsString(guid).c_str());
                                      } break;
                                      case TOutRequestState::S_WAITING_PING_SENT:
                                          s.TimePassed = 0;
                                          s.State = TOutRequestState::S_WAITING;
                                          //printf("OutReq %s PKT_PING_RESPONSE recv S_WAITING_PING_SENT -> S_WAITING\n", GetGuidAsString(guid).c_str());
                                          break;
                                      default:
                                          Y_ASSERT(0);
                                          break;
                                  }
                              }
                          }
                      } break;
                      case PKT_RESPONSE:
                      case PKT_LOCAL_RESPONSE: {
                          //printf("recv PKT_RESPONSE or PKT_LOCAL_RESPONSE\n");
                          TGUID guid;
                          reqData.Read(&guid, sizeof(guid));
                          TOutRequestHash::iterator i = OutRequests.find(guid);
                          if (i == OutRequests.end()) {
                              ; //Y_ASSERT(0); // does happen
                              //printf("OutReq %s PKT_RESPONSE recv for non-existing req\n", GetGuidAsString(guid).c_str());
                          } else {
                              FinishRequest(i, TUdpHttpResponse::OK, req);
                              //printf("OutReq %s PKT_RESPONSE recv ... -> ok\n", GetGuidAsString(guid).c_str());
                          }
                      } break;
                      case PKT_CANCEL: {
                          //printf("recv PKT_CANCEL\n");
                          TGUID guid;
                          reqData.Read(&guid, sizeof(guid));
                          TInRequestHash::iterator i = InRequests.find(guid);
                          if (i == InRequests.end()) {
                              ; //Y_ASSERT(0); // may happen
                              //printf("InReq %s PKT_CANCEL recv for non-existing req\n", GetGuidAsString(guid).c_str());
                          } else {
                              TInRequestState& s = i->second;
                              if (s.State != TInRequestState::S_CANCELED && ReportRequestCancel)
                                  UserQueues->AddCancel(guid);
                              s.State = TInRequestState::S_CANCELED;
                              //printf("InReq %s PKT_CANCEL recv\n", GetGuidAsString(guid).c_str());
                          }
                      } break;
                      case PKT_GETDEBUGINFO: {
                          //printf("recv PKT_GETDEBUGINFO\n");
                          TString dbgInfo = GetDebugInfoLocked();
                          TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
                          ms->Write(dbgInfo.c_str(), (int)dbgInfo.size());
                          SendWithHighPriority(req->Address, ms);
                      } break;
                      default:
                          Y_ASSERT(0);
                  }
              }
              // cleanup failed requests
              for (size_t k = 0; k < failedRequests.size(); ++k) {
                  const TGUID& guid = failedRequests[k];
                  TOutRequestHash::iterator i = OutRequests.find(guid);
                  if (i != OutRequests.end())
                      FinishRequest(i, TUdpHttpResponse::FAILED, nullptr, "request failed: recv no such query");
              }
          }
          void AnalyzeSendResults() {
              TSendResult res;
              while (Host->GetSendResult(&res)) {
                  //printf("Send result received\n");
                  TTransferHash::iterator k1 = TransferHash.find(res.TransferId);
                  if (k1 != TransferHash.end()) {
                      const TTransferPurpose& tp = k1->second;
                      switch (tp.Dir) {
                          case DIR_OUT: {
                              TOutRequestHash::iterator i = OutRequests.find(tp.Guid);
                              if (i != OutRequests.end()) {
                                  const TGUID& reqId = i->first;
                                  TOutRequestState& s = i->second;
                                  switch (s.State) {
                                      case TOutRequestState::S_SENDING:
                                          if (!res.Success) {
                                              FinishRequest(i, TUdpHttpResponse::FAILED, nullptr, "request failed: state S_SENDING");
                                              //printf("OutReq %s AnalyzeSendResults() S_SENDING -> failed\n", GetGuidAsString(reqId).c_str());
                                          } else {
                                              if (ReportSendRequestAcc) {
                                                  if (s.UserQueues.Get()) {
                                                      s.UserQueues->AddSendRequestAcc(reqId);
                                                  } else {
                                                      // waitable request?
                                                      TSyncRequests::iterator k2 = SyncRequests.find(reqId);
                                                      if (k2 != SyncRequests.end()) {
                                                          TIntrusivePtr<TWaitResponse>& wr = k2->second;
                                                          wr->SetRequestSent();
                                                      }
                                                  }
                                              }
                                              s.State = TOutRequestState::S_WAITING;
                                              //printf("OutReq %s AnalyzeSendResults() S_SENDING -> S_WAITING\n", GetGuidAsString(reqId).c_str());
                                              s.TimePassed = 0;
                                          }
                                          break;
                                      case TOutRequestState::S_CANCEL_AFTER_SENDING:
                                          DoSendCancel(s.Address, reqId);
                                          FinishRequest(i, TUdpHttpResponse::CANCELED, nullptr, "request failed: state S_CANCEL_AFTER_SENDING");
                                          break;
                                      case TOutRequestState::S_WAITING:
                                      case TOutRequestState::S_WAITING_PING_SENT:
                                          Y_ASSERT(0);
                                          break;
                                      case TOutRequestState::S_WAITING_PING_SENDING:
                                          Y_ASSERT(s.PingTransferId >= 0 && s.PingTransferId == res.TransferId);
                                          if (!res.Success) {
                                              FinishRequest(i, TUdpHttpResponse::FAILED, nullptr, "request failed: state S_WAITING_PING_SENDING");
                                              //printf("OutReq %s AnalyzeSendResults() S_WAITING_PING_SENDING -> failed\n", GetGuidAsString(reqId).c_str());
                                          } else {
                                              s.PingTransferId = -1;
                                              s.State = TOutRequestState::S_WAITING_PING_SENT;
                                              //printf("OutReq %s AnalyzeSendResults() S_WAITING_PING_SENDING -> S_WAITING_PING_SENT\n", GetGuidAsString(reqId).c_str());
                                              s.TimePassed = 0;
                                          }
                                          break;
                                      default:
                                          Y_ASSERT(0);
                                          break;
                                  }
                              }
                          } break;
                          case DIR_IN: {
                              TInRequestHash::iterator i = InRequests.find(tp.Guid);
                              if (i != InRequests.end()) {
                                  Y_ASSERT(i->second.State == TInRequestState::S_RESPONSE_SENDING || i->second.State == TInRequestState::S_CANCELED);
                                  InRequests.erase(i);
                                  //if (res.Success)
                                  //    printf("InReq %s AnalyzeSendResults() ... -> finished\n", GetGuidAsString(tp.Guid).c_str());
                                  //else
                                  //    printf("InReq %s AnalyzeSendResults() ... -> failed response send\n", GetGuidAsString(tp.Guid).c_str());
                              }
                          } break;
                          default:
                              Y_ASSERT(0);
                              break;
                      }
                      TransferHash.erase(k1);
                  }
              }
          }
          void SendPingsIfNeeded() {
              NHPTimer::STime tChk = PingsSendT;
              float deltaT = (float)NHPTimer::GetTimePassed(&tChk);
              if (deltaT < 0.05) {
                  return;
              }
              PingsSendT = tChk;
              deltaT = ClampVal(deltaT, 0.0f, HTTP_TIMEOUT / 3);
  
              {
                  for (TOutRequestHash::iterator i = OutRequests.begin(); i != OutRequests.end();) {
                      TOutRequestHash::iterator curIt = i++;
                      TOutRequestState& s = curIt->second;
                      const TGUID& guid = curIt->first;
                      switch (s.State) {
                          case TOutRequestState::S_WAITING:
                              s.TimePassed += deltaT;
                              if (s.TimePassed > HTTP_TIMEOUT) {
                                  TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
                                  ms->Write((char)PKT_PING);
                                  ms->Write(guid);
                                  int transId = SendWithHighPriority(s.Address, ms.Release());
                                  TransferHash[transId] = TTransferPurpose(DIR_OUT, guid);
                                  s.State = TOutRequestState::S_WAITING_PING_SENDING;
                                  //printf("OutReq %s SendPingsIfNeeded() S_WAITING -> S_WAITING_PING_SENDING\n", GetGuidAsString(guid).c_str());
                                  s.PingTransferId = transId;
                              }
                              break;
                          case TOutRequestState::S_WAITING_PING_SENT:
                              s.TimePassed += deltaT;
                              if (s.TimePassed > HTTP_TIMEOUT) {
                                  //printf("OutReq %s SendPingsIfNeeded() S_WAITING_PING_SENT -> failed\n", GetGuidAsString(guid).c_str());
                                  FinishRequest(curIt, TUdpHttpResponse::FAILED, nullptr, "request failed: http timeout in state S_WAITING_PING_SENT");
                              }
                              break;
                          default:
                              break;
                      }
                  }
              }
          }
          void Step() {
              {
                  TGuard<TSpinLock> lock(cs);
                  DoSends();
              }
              Host->Step();
              for (TIntrusivePtr<TStatsRequest> req; StatsReqList.Dequeue(&req);) {
                  switch (req->Req) {
                      case TStatsRequest::PENDING_SIZE:
                          Host->GetPendingDataSize(&req->PendingDataSize);
                          break;
                      case TStatsRequest::DEBUG_INFO: {
                          TGuard<TSpinLock> lock(cs);
                          req->DebugInfo = GetDebugInfoLocked();
                      } break;
                      case TStatsRequest::HAS_IN_REQUEST: {
                          TGuard<TSpinLock> lock(cs);
                          req->RequestFound = (InRequests.find(req->RequestId) != InRequests.end());
                      } break;
                      case TStatsRequest::GET_PEER_ADDRESS: {
                          TGuard<TSpinLock> lock(cs);
                          TInRequestHash::const_iterator i = InRequests.find(req->RequestId);
                          if (i != InRequests.end()) {
                              req->PeerAddress = i->second.Address;
                          } else {
                              TOutRequestHash::const_iterator o = OutRequests.find(req->RequestId);
                              if (o != OutRequests.end()) {
                                  req->PeerAddress = o->second.Address;
                              } else {
                                  req->PeerAddress = TUdpAddress();
                              }
                          }
                      } break;
                      case TStatsRequest::GET_PEER_QUEUE_STATS:
                          req->QueueStats = Host->GetQueueStats(req->PeerAddress);
                          break;
                      default:
                          Y_ASSERT(0);
                          break;
                  }
                  req->Complete.Signal();
              }
              {
                  TGuard<TSpinLock> lock(cs);
                  DoSends();
                  ProcessIncomingPackets();
                  AnalyzeSendResults();
                  SendPingsIfNeeded();
              }
          }
          void Wait() {
              Host->Wait(0.1f);
          }
          void DoSendCancel(const TUdpAddress& addr, const TGUID& req) {
              TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
              ms->Write((char)PKT_CANCEL);
              ms->Write(req);
              SendWithHighPriority(addr, ms);
          }
          void DoSends() {
              {
                  TBreakRequest rb;
                  while (BreakReqList.Dequeue(&rb)) {
                      InRequests.erase(rb.ReqGuid);
                  }
              }
              {
                  // cancelling requests
                  TCancelRequest rc;
                  while (CancelReqList.Dequeue(&rc)) {
                      TOutRequestHash::iterator i = OutRequests.find(rc.ReqGuid);
                      if (i == OutRequests.end()) {
                          AnticipateCancels.insert(rc.ReqGuid);
                          continue; // cancelling non existing request is ok
                      }
                      TOutRequestState& s = i->second;
                      if (s.State == TOutRequestState::S_SENDING) {
                          // we are in trouble - have not sent request and we already have to cancel it, wait send
                          s.State = TOutRequestState::S_CANCEL_AFTER_SENDING;
                      } else {
                          DoSendCancel(s.Address, rc.ReqGuid);
                          FinishRequest(i, TUdpHttpResponse::CANCELED, nullptr, "request canceled: notify requested side");
                      }
                  }
              }
              {
                  // sending replies
                  for (TSendResponse* rd = nullptr; SendRespList.Dequeue(&rd); delete rd) {
                      TInRequestHash::iterator i = InRequests.find(rd->ReqGuid);
                      if (i == InRequests.end()) {
                          Y_ASSERT(0);
                          continue;
                      }
                      TInRequestState& s = i->second;
                      if (s.State == TInRequestState::S_CANCELED) {
                          // need not send response for the canceled request
                          InRequests.erase(i);
                          continue;
                      }
  
                      Y_ASSERT(s.State == TInRequestState::S_WAITING);
                      s.State = TInRequestState::S_RESPONSE_SENDING;
                      //printf("InReq %s SendResponse() ... -> S_RESPONSE_SENDING (pkt %s)\n", GetGuidAsString(reqId).c_str(), GetGuidAsString(lowPktGuid).c_str());
  
                      TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
                      ui32 crc32 = 0;
                      int dataSize = rd->Data.ysize();
                      if (rd->Data.ysize() > MIN_SHARED_MEM_PACKET && IsLocalFast(s.Address)) {
                          TIntrusivePtr<TSharedMemory> shm = new TSharedMemory;
                          if (shm->Create(dataSize)) {
                              ms->Write((char)PKT_LOCAL_RESPONSE);
                              ms->Write(rd->ReqGuid);
                              memcpy(shm->GetPtr(), &rd->Data[0], dataSize);
                              TVector<char> empty;
                              rd->Data.swap(empty);
                              ms->AttachSharedData(shm);
                              crc32 = CalcChecksum(ms->GetChain());
                          }
                      }
                      if (ms->GetSharedData() == nullptr) {
                          ms->Write((char)PKT_RESPONSE);
                          ms->Write(rd->ReqGuid);
  
                          // to offload crc calcs from inner thread, crc of data[] is calced outside and passed in DataCrc32
                          // this means that we are calculating crc when shared memory is used
                          // it is hard to avoid since in SendResponse() we don't know if shared mem will be used (peer address is not available there)
                          TIncrementalChecksumCalcer csCalcer;
                          AddChain(&csCalcer, ms->GetChain());
                          // here we are replicating the way WriteDestructive serializes data
                          csCalcer.AddBlock(&dataSize, sizeof(dataSize));
                          csCalcer.AddBlockSum(rd->DataCrc32, dataSize);
                          crc32 = csCalcer.CalcChecksum();
  
                          ms->WriteDestructive(&rd->Data);
                          //ui32 chkCrc = CalcChecksum(ms->GetChain()); // can not use since its slow for large responses
                          //Y_ASSERT(chkCrc == crc32);
                      }
  
                      int transId = Host->Send(s.Address, ms.Release(), crc32, nullptr, rd->Priority);
                      TransferHash[transId] = TTransferPurpose(DIR_IN, rd->ReqGuid);
                  }
              }
              {
                  // sending requests
                  for (TSendRequest* rd = nullptr; SendReqList.Dequeue(&rd); delete rd) {
                      Y_ASSERT(OutRequests.find(rd->ReqGuid) == OutRequests.end());
  
                      {
                          TOutRequestState& s = OutRequests[rd->ReqGuid];
                          s.State = TOutRequestState::S_SENDING;
                          s.Address = rd->Addr;
                          s.UserQueues = rd->UserQueues;
                          //printf("OutReq %s SendRequest() ... -> S_SENDING\n", GetGuidAsString(guid).c_str());
                      }
  
                      if (rd->WR.Get())
                          SyncRequests[rd->ReqGuid] = rd->WR;
  
                      if (AnticipateCancels.find(rd->ReqGuid) != AnticipateCancels.end()) {
                          FinishRequest(OutRequests.find(rd->ReqGuid), TUdpHttpResponse::CANCELED, nullptr, "request canceled before transmitting");
                      } else {
                          TGUID pktGuid = rd->ReqGuid; // request packet id should match request id
                          int transId = Host->Send(rd->Addr, rd->Data.Release(), rd->Crc32, &pktGuid, PP_NORMAL);
                          TransferHash[transId] = TTransferPurpose(DIR_OUT, rd->ReqGuid);
                      }
                  }
              }
              if (!AnticipateCancels.empty()) {
                  AnticipateCancels.clear();
              }
          }
  
      public:
          void SendRequestImpl(const TUdpAddress& addr, const TString& url, TVector<char>* data, const TGUID& reqId,
                               TWaitResponse* wr, TRequesterUserQueues* userQueues) {
              if (data && data->size() > MAX_PACKET_SIZE) {
                  Y_VERIFY(0, "data size is too large");
              }
              //printf("SendRequest(%s)\n", url.c_str());
              if (wr)
                  wr->SetReqId(reqId);
  
              TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket;
              if (data && data->ysize() > MIN_SHARED_MEM_PACKET && IsLocalFast(addr)) {
                  int dataSize = data->ysize();
                  TIntrusivePtr<TSharedMemory> shm = new TSharedMemory;
                  if (shm->Create(dataSize)) {
                      ms->Write((char)PKT_LOCAL_REQUEST);
                      ms->WriteStroka(url);
                      memcpy(shm->GetPtr(), &(*data)[0], dataSize);
                      TVector<char> empty;
                      data->swap(empty);
                      ms->AttachSharedData(shm);
                  }
              }
              if (ms->GetSharedData() == nullptr) {
                  ms->Write((char)PKT_REQUEST);
                  ms->WriteStroka(url);
                  ms->WriteDestructive(data);
              }
  
              SendReqList.Enqueue(new TSendRequest(addr, &ms, reqId, wr, userQueues));
              Host->CancelWait();
          }
  
          void SendRequest(const TUdpAddress& addr, const TString& url, TVector<char>* data, const TGUID& reqId) override {
              SendRequestImpl(addr, url, data, reqId, nullptr, UserQueues.Get());
          }
          void CancelRequest(const TGUID& reqId) override {
              CancelReqList.Enqueue(TCancelRequest(reqId));
              Host->CancelWait();
          }
          void BreakRequest(const TGUID& reqId) override {
              BreakReqList.Enqueue(TBreakRequest(reqId));
              Host->CancelWait();
          }
  
          void SendResponseImpl(const TGUID& reqId, EPacketPriority prior, TVector<char>* data) // non-virtual, for direct call from TRequestOps
          {
              if (data && data->size() > MAX_PACKET_SIZE) {
                  Y_VERIFY(0, "data size is too large");
              }
              SendRespList.Enqueue(new TSendResponse(reqId, prior, data));
              Host->CancelWait();
          }
          void SendResponse(const TGUID& reqId, TVector<char>* data) override {
              SendResponseImpl(reqId, PP_NORMAL, data);
          }
          void SendResponseLowPriority(const TGUID& reqId, TVector<char>* data) override {
              SendResponseImpl(reqId, PP_LOW, data);
          }
          TUdpHttpRequest* GetRequest() override {
              return UserQueues->GetRequest();
          }
          TUdpHttpResponse* GetResponse() override {
              return UserQueues->GetResponse();
          }
          bool GetRequestCancel(TGUID* req) override {
              return UserQueues->GetRequestCancel(req);
          }
          bool GetSendRequestAcc(TGUID* req) override {
              return UserQueues->GetSendRequestAcc(req);
          }
          TUdpHttpResponse* Request(const TUdpAddress& addr, const TString& url, TVector<char>* data) override {
              TIntrusivePtr<TWaitResponse> wr = WaitableRequest(addr, url, data);
              wr->Wait();
              return wr->GetResponse();
          }
          TIntrusivePtr<TWaitResponse> WaitableRequest(const TUdpAddress& addr, const TString& url, TVector<char>* data) override {
              TIntrusivePtr<TWaitResponse> wr = new TWaitResponse;
              TGUID reqId;
              CreateGuid(&reqId);
              SendRequestImpl(addr, url, data, reqId, wr.Get(), nullptr);
              return wr;
          }
          TMuxEvent& GetAsyncEvent() override {
              return UserQueues->GetAsyncEvent();
          }
          int GetPort() override {
              return Socket.Get() ? Socket->GetPort() : 0;
          }
          void StopNoWait() override {
              AbortTransactions = true;
              KeepRunning = false;
              UserQueues->AsyncSignal();
              // calcel all outgoing requests
              TGuard<TSpinLock> lock(cs);
              while (!OutRequests.empty()) {
                  // cancel without informing peer that we are cancelling the request
                  FinishRequest(OutRequests.begin(), TUdpHttpResponse::CANCELED, nullptr, "request canceled: inside TUdpHttp::StopNoWait()");
              }
          }
          void ExecStatsRequest(TIntrusivePtr<TStatsRequest> req) {
              StatsReqList.Enqueue(req);
              Host->CancelWait();
              req->Complete.Wait();
          }
          TUdpAddress GetPeerAddress(const TGUID& reqId) override {
              TIntrusivePtr<TStatsRequest> req = new TStatsRequest(TStatsRequest::GET_PEER_ADDRESS);
              req->RequestId = reqId;
              ExecStatsRequest(req);
              return req->PeerAddress;
          }
          void GetPendingDataSize(TRequesterPendingDataStats* res) override {
              TIntrusivePtr<TStatsRequest> req = new TStatsRequest(TStatsRequest::PENDING_SIZE);
              ExecStatsRequest(req);
              *res = req->PendingDataSize;
          }
          bool HasRequest(const TGUID& reqId) override {
              TIntrusivePtr<TStatsRequest> req = new TStatsRequest(TStatsRequest::HAS_IN_REQUEST);
              req->RequestId = reqId;
              ExecStatsRequest(req);
              return req->RequestFound;
          }
  
      private:
          void FinishOutstandingTransactions() {
              // wait all pending requests, all new requests are canceled
              while ((!OutRequests.empty() || !InRequests.empty() || !SendRespList.IsEmpty() || !SendReqList.IsEmpty()) && !PanicAttack) {
                  while (TUdpHttpRequest* req = GetRequest()) {
                      TInRequestHash::iterator i = InRequests.find(req->ReqId);
                      //printf("dropping request(%s) (thread %d)\n", req->Url.c_str(), ThreadId());
                      delete req;
                      if (i == InRequests.end()) {
                          Y_ASSERT(0);
                          continue;
                      }
                      InRequests.erase(i);
                  }
                  Step();
                  sleep(0);
              }
          }
          static void* ExecServerThread(void* param) {
              BindToSocket(0);
              SetHighestThreadPriority();
              TUdpHttp* pThis = (TUdpHttp*)param;
              pThis->Host = CreateUdpHost(pThis->Socket);
              pThis->HasStarted.Signal();
              if (!pThis->Host) {
                  pThis->Socket.Drop();
                  return nullptr;
              }
              NHPTimer::GetTime(&pThis->PingsSendT);
              while (pThis->KeepRunning && !PanicAttack) {
                  if (HeartbeatTimeout > 0) {
                      NHPTimer::STime chk = LastHeartbeat;
                      double passed = NHPTimer::GetTimePassed(&chk);
                      if (passed > HeartbeatTimeout) {
                          StopAllNetLibaThreads();
                          fprintf(stderr, "%s\tTUdpHttp\tWaiting for %0.2f, time limit %0.2f, commit a suicide!11\n", Now().ToStringUpToSeconds().c_str(), passed, HeartbeatTimeout);
                          fflush(stderr);
  #ifndef _win_
                          killpg(0, SIGKILL);
  #endif
                          abort();
                          break;
                      }
                  }
                  pThis->Step();
                  pThis->Wait();
              }
              if (!pThis->AbortTransactions && !PanicAttack)
                  pThis->FinishOutstandingTransactions();
              pThis->Host = nullptr;
              return nullptr;
          }
          ~TUdpHttp() override {
              if (myThread.Running()) {
                  KeepRunning = false;
                  myThread.Join();
              }
              for (TIntrusivePtr<TStatsRequest> req; StatsReqList.Dequeue(&req);) {
                  req->Complete.Signal();
              }
          }
  
      public:
          TUdpHttp()
              : myThread(TThread::TParams(ExecServerThread, (void*)this).SetName("nl6_udp_host"))
              , KeepRunning(true)
              , AbortTransactions(false)
              , PingsSendT(0)
              , ReportRequestCancel(false)
              , ReportSendRequestAcc(false)
          {
              NHPTimer::GetTime(&PingsSendT);
              QueueSizes = new TRequesterUserQueueSizes;
              UserQueues = new TRequesterUserQueues(QueueSizes.Get());
          }
          bool Start(const TIntrusivePtr<NNetlibaSocket::ISocket>& socket) {
              Y_ASSERT(Host.Get() == nullptr);
              Socket = socket;
              myThread.Start();
              HasStarted.Wait();
  
              if (Host.Get()) {
                  return true;
              }
              Socket.Drop();
              return false;
          }
          TString GetDebugInfoLocked() {
              TString res = KeepRunning ? "State: running\n" : "State: stopping\n";
              res += Host->GetDebugInfo();
  
              char buf[1000];
              TRequesterUserQueueSizes* qs = QueueSizes.Get();
              sprintf(buf, "\nRequest queue %d (%d bytes)\n", (int)qs->ReqCount, (int)qs->ReqQueueSize);
              res += buf;
              sprintf(buf, "Response queue %d (%d bytes)\n", (int)qs->RespCount, (int)qs->RespQueueSize);
              res += buf;
  
              const char* outReqStateNames[] = {
                  "S_SENDING",
                  "S_WAITING",
                  "S_WAITING_PING_SENDING",
                  "S_WAITING_PING_SENT",
                  "S_CANCEL_AFTER_SENDING"};
              const char* inReqStateNames[] = {
                  "S_WAITING",
                  "S_RESPONSE_SENDING",
                  "S_CANCELED"};
              res += "\nOut requests:\n";
              for (TOutRequestHash::const_iterator i = OutRequests.begin(); i != OutRequests.end(); ++i) {
                  const TGUID& gg = i->first;
                  const TOutRequestState& s = i->second;
                  bool isSync = SyncRequests.find(gg) != SyncRequests.end();
                  sprintf(buf, "%s\t%s  %s  TimePassed: %g  %s\n",
                          GetAddressAsString(s.Address).c_str(), GetGuidAsString(gg).c_str(), outReqStateNames[s.State],
                          s.TimePassed * 1000,
                          isSync ? "isSync" : "");
                  res += buf;
              }
              res += "\nIn requests:\n";
              for (TInRequestHash::const_iterator i = InRequests.begin(); i != InRequests.end(); ++i) {
                  const TGUID& gg = i->first;
                  const TInRequestState& s = i->second;
                  sprintf(buf, "%s\t%s  %s\n",
                          GetAddressAsString(s.Address).c_str(), GetGuidAsString(gg).c_str(), inReqStateNames[s.State]);
                  res += buf;
              }
              return res;
          }
          TString GetDebugInfo() override {
              TIntrusivePtr<TStatsRequest> req = new TStatsRequest(TStatsRequest::DEBUG_INFO);
              ExecStatsRequest(req);
              return req->DebugInfo;
          }
          void GetRequestQueueSize(TRequesterQueueStats* res) override {
              TRequesterUserQueueSizes* qs = QueueSizes.Get();
              res->ReqCount = (int)qs->ReqCount;
              res->RespCount = (int)qs->RespCount;
              res->ReqQueueSize = (int)qs->ReqQueueSize;
              res->RespQueueSize = (int)qs->RespQueueSize;
          }
          TRequesterUserQueueSizes* GetQueueSizes() const {
              return QueueSizes.Get();
          }
          IRequestOps* CreateSubRequester() override;
          void EnableReportRequestCancel() override {
              ReportRequestCancel = true;
          }
          void EnableReportSendRequestAcc() override {
              ReportSendRequestAcc = true;
          }
          TIntrusivePtr<IPeerQueueStats> GetQueueStats(const TUdpAddress& addr) override {
              TIntrusivePtr<TStatsRequest> req = new TStatsRequest(TStatsRequest::GET_PEER_QUEUE_STATS);
              req->PeerAddress = addr;
              ExecStatsRequest(req);
              return req->QueueStats;
          }
      };
  
      //////////////////////////////////////////////////////////////////////////
      static void ReadShm(TSharedMemory* shm, TVector<char>* data) {
          Y_ASSERT(shm);
          int dataSize = shm->GetSize();
          data->yresize(dataSize);
          memcpy(&(*data)[0], shm->GetPtr(), dataSize);
      }
  
      static void LoadRequestData(TUdpHttpRequest* res) {
          if (!res)
              return;
          {
              TBlockChainIterator reqData(res->DataHolder->Data->GetChain());
              char pktType;
              reqData.Read(&pktType, 1);
              ReadArr(&reqData, &res->Url);
              if (pktType == PKT_REQUEST) {
                  ReadYArr(&reqData, &res->Data);
              } else if (pktType == PKT_LOCAL_REQUEST) {
                  ReadShm(res->DataHolder->Data->GetSharedData(), &res->Data);
              } else
                  Y_ASSERT(0);
              if (reqData.HasFailed()) {
                  Y_ASSERT(0 && "wrong format, memory corruption suspected");
                  res->Url = "";
                  res->Data.clear();
              }
          }
          res->DataHolder.Reset(nullptr);
      }
  
      static void LoadResponseData(TUdpHttpResponse* res) {
          if (!res || res->DataHolder.Get() == nullptr)
              return;
          {
              TBlockChainIterator reqData(res->DataHolder->Data->GetChain());
              char pktType;
              reqData.Read(&pktType, 1);
              TGUID guid;
              reqData.Read(&guid, sizeof(guid));
              Y_ASSERT(res->ReqId == guid);
              if (pktType == PKT_RESPONSE) {
                  ReadYArr(&reqData, &res->Data);
              } else if (pktType == PKT_LOCAL_RESPONSE) {
                  ReadShm(res->DataHolder->Data->GetSharedData(), &res->Data);
              } else
                  Y_ASSERT(0);
              if (reqData.HasFailed()) {
                  Y_ASSERT(0 && "wrong format, memory corruption suspected");
                  res->Ok = TUdpHttpResponse::FAILED;
                  res->Data.clear();
                  res->Error = "wrong response format";
              }
          }
          res->DataHolder.Reset(nullptr);
      }
  
      //////////////////////////////////////////////////////////////////////////
      // IRequestOps::TWaitResponse
      TUdpHttpResponse* IRequestOps::TWaitResponse::GetResponse() {
          if (!Response)
              return nullptr;
          TUdpHttpResponse* res = Response;
          Response = nullptr;
          LoadResponseData(res);
          return res;
      }
  
      void IRequestOps::TWaitResponse::SetResponse(TUdpHttpResponse* r) {
          Y_ASSERT(Response == nullptr || r == nullptr);
          if (r)
              Response = r;
          CompleteEvent.Signal();
      }
  
      //////////////////////////////////////////////////////////////////////////
      // TRequesterUserQueues
      TUdpHttpRequest* TRequesterUserQueues::GetRequest() {
          TUdpHttpRequest* res = nullptr;
          ReqList.Dequeue(&res);
          if (res) {
              AtomicAdd(QueueSizes->ReqCount, -1);
              AtomicAdd(QueueSizes->ReqQueueSize, -GetPacketSize(res->DataHolder.Get()));
          }
          UpdateAsyncSignalState();
          LoadRequestData(res);
          return res;
      }
  
      TUdpHttpResponse* TRequesterUserQueues::GetResponse() {
          TUdpHttpResponse* res = nullptr;
          ResponseList.Dequeue(&res);
          if (res) {
              AtomicAdd(QueueSizes->RespCount, -1);
              AtomicAdd(QueueSizes->RespQueueSize, -GetPacketSize(res->DataHolder.Get()));
          }
          UpdateAsyncSignalState();
          LoadResponseData(res);
          return res;
      }
  
      //////////////////////////////////////////////////////////////////////////
      class TRequestOps: public IRequestOps {
          TIntrusivePtr<TUdpHttp> Requester;
          TIntrusivePtr<TRequesterUserQueues> UserQueues;
  
      public:
          TRequestOps(TUdpHttp* req)
              : Requester(req)
          {
              UserQueues = new TRequesterUserQueues(req->GetQueueSizes());
          }
          void SendRequest(const TUdpAddress& addr, const TString& url, TVector<char>* data, const TGUID& reqId) override {
              Requester->SendRequestImpl(addr, url, data, reqId, nullptr, UserQueues.Get());
          }
          void CancelRequest(const TGUID& reqId) override {
              Requester->CancelRequest(reqId);
          }
          void BreakRequest(const TGUID& reqId) override {
              Requester->BreakRequest(reqId);
          }
  
          void SendResponse(const TGUID& reqId, TVector<char>* data) override {
              Requester->SendResponseImpl(reqId, PP_NORMAL, data);
          }
          void SendResponseLowPriority(const TGUID& reqId, TVector<char>* data) override {
              Requester->SendResponseImpl(reqId, PP_LOW, data);
          }
          TUdpHttpRequest* GetRequest() override {
              Y_ASSERT(0);
              //return UserQueues.GetRequest();
              return nullptr; // all requests are routed to the main requester
          }
          TUdpHttpResponse* GetResponse() override {
              return UserQueues->GetResponse();
          }
          bool GetRequestCancel(TGUID*) override {
              Y_ASSERT(0);
              return false; // all request cancels are routed to the main requester
          }
          bool GetSendRequestAcc(TGUID* req) override {
              return UserQueues->GetSendRequestAcc(req);
          }
          // sync mode
          TUdpHttpResponse* Request(const TUdpAddress& addr, const TString& url, TVector<char>* data) override {
              return Requester->Request(addr, url, data);
          }
          TIntrusivePtr<TWaitResponse> WaitableRequest(const TUdpAddress& addr, const TString& url, TVector<char>* data) override {
              return Requester->WaitableRequest(addr, url, data);
          }
          //
          TMuxEvent& GetAsyncEvent() override {
              return UserQueues->GetAsyncEvent();
          }
      };
  
      IRequestOps* TUdpHttp::CreateSubRequester() {
          return new TRequestOps(this);
      }
  
      //////////////////////////////////////////////////////////////////////////
      void AbortOnFailedRequest(TUdpHttpResponse* answer) {
          if (answer && answer->Ok == TUdpHttpResponse::FAILED) {
              fprintf(stderr, "Failed request to host %s\n", GetAddressAsString(answer->PeerAddress).data());
              fprintf(stderr, "Error description: %s\n", answer->Error.data());
              fflush(nullptr);
              Y_ASSERT(0);
              abort();
          }
      }
  
      TString GetDebugInfo(const TUdpAddress& addr, double timeout) {
          NHPTimer::STime start;
          NHPTimer::GetTime(&start);
          TIntrusivePtr<IUdpHost> host = CreateUdpHost(0);
          {
              TAutoPtr<TRopeDataPacket> rq = new TRopeDataPacket;
              rq->Write((char)PKT_GETDEBUGINFO);
              ui32 crc32 = CalcChecksum(rq->GetChain());
              host->Send(addr, rq.Release(), crc32, nullptr, PP_HIGH);
          }
          for (;;) {
              TAutoPtr<TRequest> ptr = host->GetRequest();
              if (ptr.Get()) {
                  TBlockChainIterator reqData(ptr->Data->GetChain());
                  int sz = reqData.GetSize();
                  TString res;
                  res.resize(sz);
                  reqData.Read(res.begin(), sz);
                  return res;
              }
              host->Step();
              host->Wait(0.1f);
  
              NHPTimer::STime now;
              NHPTimer::GetTime(&now);
              if (NHPTimer::GetSeconds(now - start) > timeout) {
                  return TString();
              }
          }
      }
  
      void Kill(const TUdpAddress& addr) {
          TIntrusivePtr<IUdpHost> host = CreateUdpHost(0);
          host->Kill(addr);
      }
  
      void StopAllNetLibaThreads() {
          PanicAttack = true; // AAAA!!!!
      }
  
      void SetNetLibaHeartbeatTimeout(double timeoutSec) {
          NetLibaHeartbeat();
          HeartbeatTimeout = timeoutSec;
      }
  
      void NetLibaHeartbeat() {
          NHPTimer::STime now;
          NHPTimer::GetTime(&now);
          LastHeartbeat = now;
      }
  
      IRequester* CreateHttpUdpRequester(int port) {
          if (PanicAttack)
              return nullptr;
  
          TIntrusivePtr<NNetlibaSocket::ISocket> socket = NNetlibaSocket::CreateSocket();
          socket->Open(port);
          if (!socket->IsValid())
              return nullptr;
  
          return CreateHttpUdpRequester(socket);
      }
  
      IRequester* CreateHttpUdpRequester(const TIntrusivePtr<NNetlibaSocket::ISocket>& socket) {
          if (PanicAttack)
              return nullptr;
  
          TIntrusivePtr<TUdpHttp> res(new TUdpHttp);
          if (!res->Start(socket))
              return nullptr;
          return res.Release();
      }
  
  }
  