src/ipc/UdsOp.cc

   1 /*
   2  * Copyright (C) 1996-2023 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 54    Interprocess Communication */
  10
  11 #include "squid.h"
  12 #include "base/TextException.h"
  13 #include "comm.h"
  14 #include "comm/Connection.h"
  15 #include "comm/Write.h"
  16 #include "CommCalls.h"
  17 #include "ipc/UdsOp.h"
  18
  19 Ipc::UdsOp::UdsOp(const String& pathAddr):
  20     AsyncJob("Ipc::UdsOp"),
  21     address(PathToAddress(pathAddr)),
  22     options(COMM_NONBLOCKING)
  23 {
  24     debugs(54, 5, '[' << this << "] pathAddr=" << pathAddr);
  25 }
  26
  27 Ipc::UdsOp::~UdsOp()
  28 {
  29     debugs(54, 5, '[' << this << ']');
  30     if (Comm::IsConnOpen(conn_))
  31         conn_->close();
  32     conn_ = nullptr;
  33 }
  34
  35 void Ipc::UdsOp::setOptions(int newOptions)
  36 {
  37     options = newOptions;
  38 }
  39
  40 Comm::ConnectionPointer &
  41 Ipc::UdsOp::conn()
  42 {
  43     if (!Comm::IsConnOpen(conn_)) {
  44         if (options & COMM_DOBIND)
  45             unlink(address.sun_path);
  46         if (conn_ == nullptr)
  47             conn_ = new Comm::Connection;
  48         conn_->fd = comm_open_uds(SOCK_DGRAM, 0, &address, options);
  49         Must(Comm::IsConnOpen(conn_));
  50     }
  51     return conn_;
  52 }
  53
  54 void Ipc::UdsOp::setTimeout(int seconds, const char *handlerName)
  55 {
  56     typedef CommCbMemFunT<UdsOp, CommTimeoutCbParams> Dialer;
  57     AsyncCall::Pointer handler = asyncCall(54,5, handlerName,
  58                                            Dialer(CbcPointer<UdsOp>(this), &UdsOp::noteTimeout));
  59     commSetConnTimeout(conn(), seconds, handler);
  60 }
  61
  62 void Ipc::UdsOp::clearTimeout()
  63 {
  64     commUnsetConnTimeout(conn());
  65 }
  66
  67 void Ipc::UdsOp::noteTimeout(const CommTimeoutCbParams &)
  68 {
  69     timedout(); // our kid handles communication timeout
  70 }
  71
  72 struct sockaddr_un
  73 Ipc::PathToAddress(const String& pathAddr) {
  74     assert(pathAddr.size() != 0);
  75     struct sockaddr_un unixAddr;
  76     memset(&unixAddr, 0, sizeof(unixAddr));
  77     unixAddr.sun_family = AF_LOCAL;
  78     xstrncpy(unixAddr.sun_path, pathAddr.termedBuf(), sizeof(unixAddr.sun_path));
  79     return unixAddr;
  80 }
  81
  82 CBDATA_NAMESPACED_CLASS_INIT(Ipc, UdsSender);
  83
  84 Ipc::UdsSender::UdsSender(const String& pathAddr, const TypedMsgHdr& aMessage):
  85     UdsOp(pathAddr),
  86     codeContext(CodeContext::Current()),
  87     message(aMessage),
  88     retries(10), // TODO: make configurable?
  89     timeout(10), // TODO: make configurable?
  90     sleeping(false),
  91     writing(false)
  92 {
  93     message.address(address);
  94 }
  95
  96 void Ipc::UdsSender::swanSong()
  97 {
  98     // did we abort while waiting between retries?
  99     if (sleeping)
 100         cancelSleep();
 101
 102     UdsOp::swanSong();
 103 }
 104
 105 void Ipc::UdsSender::start()
 106 {
 107     UdsOp::start();
 108     write();
 109     if (timeout > 0)
 110         setTimeout(timeout, "Ipc::UdsSender::noteTimeout");
 111 }
 112
 113 bool Ipc::UdsSender::doneAll() const
 114 {
 115     return !writing && !sleeping && UdsOp::doneAll();
 116 }
 117
 118 void Ipc::UdsSender::write()
 119 {
 120     debugs(54, 5, MYNAME);
 121     typedef CommCbMemFunT<UdsSender, CommIoCbParams> Dialer;
 122     AsyncCall::Pointer writeHandler = JobCallback(54, 5,
 123                                       Dialer, this, UdsSender::wrote);
 124     Comm::Write(conn(), message.raw(), message.size(), writeHandler, nullptr);
 125     writing = true;
 126 }
 127
 128 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 129 {
 130     debugs(54, 5, params.conn << " flag " << params.flag << " retries " << retries << " [" << this << ']');
 131     writing = false;
 132     if (params.flag != Comm::OK && retries-- > 0) {
 133         // perhaps a fresh connection and more time will help?
 134         conn()->close();
 135         startSleep();
 136     }
 137 }
 138
 139 /// pause for a while before resending the message
 140 void Ipc::UdsSender::startSleep()
 141 {
 142     Must(!sleeping);
 143     sleeping = true;
 144     eventAdd("Ipc::UdsSender::DelayedRetry",
 145              Ipc::UdsSender::DelayedRetry,
 146              new Pointer(this), 1, 0, false); // TODO: Use Fibonacci increments
 147 }
 148
 149 /// stop sleeping (or do nothing if we were not)
 150 void Ipc::UdsSender::cancelSleep()
 151 {
 152     if (sleeping) {
 153         // Why not delete the event? See Comm::ConnOpener::cancelSleep().
 154         sleeping = false;
 155         debugs(54, 9, "stops sleeping");
 156     }
 157 }
 158
 159 /// legacy wrapper for Ipc::UdsSender::delayedRetry()
 160 void Ipc::UdsSender::DelayedRetry(void *data)
 161 {
 162     Pointer *ptr = static_cast<Pointer*>(data);
 163     assert(ptr);
 164     if (UdsSender *us = dynamic_cast<UdsSender*>(ptr->valid())) {
 165         CallBack(us->codeContext, [&us] {
 166             CallJobHere(54, 4, us, UdsSender, delayedRetry);
 167         });
 168     }
 169     delete ptr;
 170 }
 171
 172 /// make another sending attempt after a pause
 173 void Ipc::UdsSender::delayedRetry()
 174 {
 175     debugs(54, 5, sleeping);
 176     if (sleeping) {
 177         sleeping = false;
 178         write(); // reopens the connection if needed
 179     }
 180 }
 181
 182 void Ipc::UdsSender::timedout()
 183 {
 184     debugs(54, 5, MYNAME);
 185     mustStop("timedout");
 186 }
 187
 188 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 189 {
 190     AsyncJob::Start(new UdsSender(toAddress, message));
 191 }
 192
 193 const Comm::ConnectionPointer &
 194 Ipc::ImportFdIntoComm(const Comm::ConnectionPointer &conn, int socktype, int protocol, Ipc::FdNoteId noteId)
 195 {
 196     struct sockaddr_storage addr;
 197     socklen_t len = sizeof(addr);
 198     if (getsockname(conn->fd, reinterpret_cast<sockaddr*>(&addr), &len) == 0) {
 199         conn->remote = addr;
 200         struct addrinfo* addr_info = nullptr;
 201         conn->remote.getAddrInfo(addr_info);
 202         addr_info->ai_socktype = socktype;
 203         addr_info->ai_protocol = protocol;
 204         comm_import_opened(conn, Ipc::FdNote(noteId), addr_info);
 205         Ip::Address::FreeAddr(addr_info);
 206     } else {
 207         int xerrno = errno;
 208         debugs(54, DBG_CRITICAL, "ERROR: Ipc::ImportFdIntoComm: " << conn << ' ' << xstrerr(xerrno));
 209         conn->close();
 210     }
 211     return conn;
 212 }
 213