src/ipc/UdsOp.cc

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