src/ipc/UdsOp.cc

   1 /*
   2  * Copyright (C) 1996-2014 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, HERE << '[' << this << "] pathAddr=" << pathAddr);
  25 }
  26
  27 Ipc::UdsOp::~UdsOp()
  28 {
  29     debugs(54, 5, HERE << '[' << this << ']');
  30     if (Comm::IsConnOpen(conn_))
  31         conn_->close();
  32     conn_ = NULL;
  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_ == NULL)
  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     message(aMessage),
  87     retries(10), // TODO: make configurable?
  88     timeout(10), // TODO: make configurable?
  89     sleeping(false),
  90     writing(false)
  91 {
  92     message.address(address);
  93 }
  94
  95 void Ipc::UdsSender::swanSong()
  96 {
  97     // did we abort while waiting between retries?
  98     if (sleeping)
  99         cancelSleep();
 100
 101     UdsOp::swanSong();
 102 }
 103
 104 void Ipc::UdsSender::start()
 105 {
 106     UdsOp::start();
 107     write();
 108     if (timeout > 0)
 109         setTimeout(timeout, "Ipc::UdsSender::noteTimeout");
 110 }
 111
 112 bool Ipc::UdsSender::doneAll() const
 113 {
 114     return !writing && !sleeping && UdsOp::doneAll();
 115 }
 116
 117 void Ipc::UdsSender::write()
 118 {
 119     debugs(54, 5, HERE);
 120     typedef CommCbMemFunT<UdsSender, CommIoCbParams> Dialer;
 121     AsyncCall::Pointer writeHandler = JobCallback(54, 5,
 122                                       Dialer, this, UdsSender::wrote);
 123     Comm::Write(conn(), message.raw(), message.size(), writeHandler, NULL);
 124     writing = true;
 125 }
 126
 127 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 128 {
 129     debugs(54, 5, HERE << params.conn << " flag " << params.flag << " retries " << retries << " [" << this << ']');
 130     writing = false;
 131     if (params.flag != Comm::OK && retries-- > 0) {
 132         // perhaps a fresh connection and more time will help?
 133         conn()->close();
 134         startSleep();
 135     }
 136 }
 137
 138 /// pause for a while before resending the message
 139 void Ipc::UdsSender::startSleep()
 140 {
 141     Must(!sleeping);
 142     sleeping = true;
 143     eventAdd("Ipc::UdsSender::DelayedRetry",
 144              Ipc::UdsSender::DelayedRetry,
 145              new Pointer(this), 1, 0, false); // TODO: Use Fibonacci increments
 146 }
 147
 148 /// stop sleeping (or do nothing if we were not)
 149 void Ipc::UdsSender::cancelSleep()
 150 {
 151     if (sleeping) {
 152         // Why not delete the event? See Comm::ConnOpener::cancelSleep().
 153         sleeping = false;
 154         debugs(54, 9, "stops sleeping");
 155     }
 156 }
 157
 158 /// legacy wrapper for Ipc::UdsSender::delayedRetry()
 159 void Ipc::UdsSender::DelayedRetry(void *data)
 160 {
 161     Pointer *ptr = static_cast<Pointer*>(data);
 162     assert(ptr);
 163     if (UdsSender *us = dynamic_cast<UdsSender*>(ptr->valid())) {
 164         // get back inside AsyncJob protection by scheduling an async job call
 165         typedef NullaryMemFunT<Ipc::UdsSender> Dialer;
 166         AsyncCall::Pointer call = JobCallback(54, 4, Dialer, us, Ipc::UdsSender::delayedRetry);
 167         ScheduleCallHere(call);
 168     }
 169     delete ptr;
 170 }
 171
 172 /// make another sending attempt after a pause
 173 void Ipc::UdsSender::delayedRetry()
 174 {
 175     debugs(54, 5, HERE << 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, HERE);
 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_in 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 = NULL;
 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         debugs(54, DBG_CRITICAL, "ERROR: Ipc::ImportFdIntoComm: " << conn << ' ' << xstrerror());
 208         conn->close();
 209     }
 210     return conn;
 211 }
 212