src/ipc/UdsOp.cc

   1 /*
   2  * DEBUG: section 54    Interprocess Communication
   3  *
   4  */
   5
   6 #include "squid.h"
   7 #include "comm.h"
   8 #include "CommCalls.h"
   9 #include "comm/Connection.h"
  10 #include "comm/Write.h"
  11 #include "base/TextException.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         writing(false)
  85 {
  86     message.address(address);
  87 }
  88
  89 void Ipc::UdsSender::start()
  90 {
  91     UdsOp::start();
  92     write();
  93     if (timeout > 0)
  94         setTimeout(timeout, "Ipc::UdsSender::noteTimeout");
  95 }
  96
  97 bool Ipc::UdsSender::doneAll() const
  98 {
  99     return !writing && UdsOp::doneAll();
 100 }
 101
 102 void Ipc::UdsSender::write()
 103 {
 104     debugs(54, 5, HERE);
 105     typedef CommCbMemFunT<UdsSender, CommIoCbParams> Dialer;
 106     AsyncCall::Pointer writeHandler = JobCallback(54, 5,
 107                                       Dialer, this, UdsSender::wrote);
 108     Comm::Write(conn(), message.raw(), message.size(), writeHandler, NULL);
 109     writing = true;
 110 }
 111
 112 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 113 {
 114     debugs(54, 5, HERE << params.conn << " flag " << params.flag << " retries " << retries << " [" << this << ']');
 115     writing = false;
 116     if (params.flag != COMM_OK && retries-- > 0) {
 117         sleep(1); // do not spend all tries at once; XXX: use an async timed event instead of blocking here; store the time when we started writing so that we do not sleep if not needed?
 118         write(); // XXX: should we close on error so that conn() reopens?
 119     }
 120 }
 121
 122 void Ipc::UdsSender::timedout()
 123 {
 124     debugs(54, 5, HERE);
 125     mustStop("timedout");
 126 }
 127
 128 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 129 {
 130     AsyncJob::Start(new UdsSender(toAddress, message));
 131 }
 132
 133 const Comm::ConnectionPointer &
 134 Ipc::ImportFdIntoComm(const Comm::ConnectionPointer &conn, int socktype, int protocol, Ipc::FdNoteId noteId)
 135 {
 136     struct sockaddr_in addr;
 137     socklen_t len = sizeof(addr);
 138     if (getsockname(conn->fd, reinterpret_cast<sockaddr*>(&addr), &len) == 0) {
 139         conn->remote = addr;
 140         struct addrinfo* addr_info = NULL;
 141         conn->remote.GetAddrInfo(addr_info);
 142         addr_info->ai_socktype = socktype;
 143         addr_info->ai_protocol = protocol;
 144         comm_import_opened(conn, Ipc::FdNote(noteId), addr_info);
 145         conn->remote.FreeAddrInfo(addr_info);
 146     } else {
 147         debugs(54, DBG_CRITICAL, "ERROR: Ipc::ImportFdIntoComm: " << conn << ' ' << xstrerror());
 148         conn->close();
 149     }
 150     return conn;
 151 }