src/ipc/UdsOp.cc

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