src/ipc/UdsOp.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 54    Interprocess Communication
   5  *
   6  */
   7
   8 #include "squid.h"
   9 #include "comm.h"
  10 #include "CommCalls.h"
  11 #include "comm/Connection.h"
  12 #include "comm/Write.h"
  13 #include "base/TextException.h"
  14 #include "ipc/UdsOp.h"
  15
  16 Ipc::UdsOp::UdsOp(const String& pathAddr):
  17         AsyncJob("Ipc::UdsOp"),
  18         address(PathToAddress(pathAddr)),
  19         options(COMM_NONBLOCKING)
  20 {
  21     debugs(54, 5, HERE << '[' << this << "] pathAddr=" << pathAddr);
  22 }
  23
  24 Ipc::UdsOp::~UdsOp()
  25 {
  26     debugs(54, 5, HERE << '[' << this << ']');
  27     if (Comm::IsConnOpen(conn_))
  28         conn_->close();
  29     conn_ = NULL;
  30 }
  31
  32 void Ipc::UdsOp::setOptions(int newOptions)
  33 {
  34     options = newOptions;
  35 }
  36
  37 Comm::ConnectionPointer &
  38 Ipc::UdsOp::conn()
  39 {
  40     if (!Comm::IsConnOpen(conn_)) {
  41         if (options & COMM_DOBIND)
  42             unlink(address.sun_path);
  43         if (conn_ == NULL)
  44             conn_ = new Comm::Connection;
  45         conn_->fd = comm_open_uds(SOCK_DGRAM, 0, &address, options);
  46         Must(Comm::IsConnOpen(conn_));
  47     }
  48     return conn_;
  49 }
  50
  51 void Ipc::UdsOp::setTimeout(int seconds, const char *handlerName)
  52 {
  53     typedef CommCbMemFunT<UdsOp, CommTimeoutCbParams> Dialer;
  54     AsyncCall::Pointer handler = asyncCall(54,5, handlerName,
  55                                            Dialer(CbcPointer<UdsOp>(this), &UdsOp::noteTimeout));
  56     commSetConnTimeout(conn(), seconds, handler);
  57 }
  58
  59 void Ipc::UdsOp::clearTimeout()
  60 {
  61     commUnsetConnTimeout(conn());
  62 }
  63
  64 void Ipc::UdsOp::noteTimeout(const CommTimeoutCbParams &)
  65 {
  66     timedout(); // our kid handles communication timeout
  67 }
  68
  69 struct sockaddr_un
  70 Ipc::PathToAddress(const String& pathAddr) {
  71     assert(pathAddr.size() != 0);
  72     struct sockaddr_un unixAddr;
  73     memset(&unixAddr, 0, sizeof(unixAddr));
  74     unixAddr.sun_family = AF_LOCAL;
  75     xstrncpy(unixAddr.sun_path, pathAddr.termedBuf(), sizeof(unixAddr.sun_path));
  76     return unixAddr;
  77 }
  78
  79 CBDATA_NAMESPACED_CLASS_INIT(Ipc, UdsSender);
  80
  81 Ipc::UdsSender::UdsSender(const String& pathAddr, const TypedMsgHdr& aMessage):
  82         UdsOp(pathAddr),
  83         message(aMessage),
  84         retries(10), // TODO: make configurable?
  85         timeout(10), // TODO: make configurable?
  86         writing(false)
  87 {
  88     message.address(address);
  89 }
  90
  91 void Ipc::UdsSender::start()
  92 {
  93     UdsOp::start();
  94     write();
  95     if (timeout > 0)
  96         setTimeout(timeout, "Ipc::UdsSender::noteTimeout");
  97 }
  98
  99 bool Ipc::UdsSender::doneAll() const
 100 {
 101     return !writing && UdsOp::doneAll();
 102 }
 103
 104 void Ipc::UdsSender::write()
 105 {
 106     debugs(54, 5, HERE);
 107     typedef CommCbMemFunT<UdsSender, CommIoCbParams> Dialer;
 108     AsyncCall::Pointer writeHandler = JobCallback(54, 5,
 109                                       Dialer, this, UdsSender::wrote);
 110     Comm::Write(conn(), message.raw(), message.size(), writeHandler, NULL);
 111     writing = true;
 112 }
 113
 114 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 115 {
 116     debugs(54, 5, HERE << params.conn << " flag " << params.flag << " retries " << retries << " [" << this << ']');
 117     writing = false;
 118     if (params.flag != COMM_OK && retries-- > 0) {
 119         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?
 120         write(); // XXX: should we close on error so that conn() reopens?
 121     }
 122 }
 123
 124 void Ipc::UdsSender::timedout()
 125 {
 126     debugs(54, 5, HERE);
 127     mustStop("timedout");
 128 }
 129
 130 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 131 {
 132     AsyncJob::Start(new UdsSender(toAddress, message));
 133 }
 134
 135 const Comm::ConnectionPointer &
 136 Ipc::ImportFdIntoComm(const Comm::ConnectionPointer &conn, int socktype, int protocol, Ipc::FdNoteId noteId)
 137 {
 138     struct sockaddr_in addr;
 139     socklen_t len = sizeof(addr);
 140     if (getsockname(conn->fd, reinterpret_cast<sockaddr*>(&addr), &len) == 0) {
 141         conn->remote = addr;
 142         struct addrinfo* addr_info = NULL;
 143         conn->remote.GetAddrInfo(addr_info);
 144         addr_info->ai_socktype = socktype;
 145         addr_info->ai_protocol = protocol;
 146         comm_import_opened(conn, Ipc::FdNote(noteId), addr_info);
 147         conn->remote.FreeAddrInfo(addr_info);
 148     } else {
 149         debugs(54, DBG_CRITICAL, "ERROR: Ipc::ImportFdIntoComm: " << conn << ' ' << xstrerror());
 150         conn->close();
 151     }
 152     return conn;
 153 }