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/Write.h"
  13 #include "base/TextException.h"
  14 #include "ipc/UdsOp.h"
  15
  16
  17 Ipc::UdsOp::UdsOp(const String& pathAddr):
  18         AsyncJob("Ipc::UdsOp"),
  19         address(PathToAddress(pathAddr)),
  20         options(COMM_NONBLOCKING),
  21         fd_(-1)
  22 {
  23     debugs(54, 5, HERE << '[' << this << "] pathAddr=" << pathAddr);
  24 }
  25
  26 Ipc::UdsOp::~UdsOp()
  27 {
  28     debugs(54, 5, HERE << '[' << this << ']');
  29     if (fd_ >= 0)
  30         comm_close(fd_);
  31 }
  32
  33 void Ipc::UdsOp::setOptions(int newOptions)
  34 {
  35     options = newOptions;
  36 }
  37
  38 int Ipc::UdsOp::fd()
  39 {
  40     if (fd_ < 0) {
  41         if (options & COMM_DOBIND)
  42             unlink(address.sun_path);
  43         fd_ = comm_open_uds(SOCK_DGRAM, 0, &address, options);
  44         Must(fd_ >= 0);
  45     }
  46     return fd_;
  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     commSetTimeout(fd(), seconds, handler);
  55 }
  56
  57 void Ipc::UdsOp::clearTimeout()
  58 {
  59     commSetTimeout(fd(), -1, NULL, NULL); // TODO: add Comm::ClearTimeout(fd)
  60 }
  61
  62 void Ipc::UdsOp::noteTimeout(const CommTimeoutCbParams &)
  63 {
  64     timedout(); // our kid handles communication timeout
  65 }
  66
  67
  68 struct sockaddr_un
  69 Ipc::PathToAddress(const String& pathAddr) {
  70     assert(pathAddr.size() != 0);
  71     struct sockaddr_un unixAddr;
  72     memset(&unixAddr, 0, sizeof(unixAddr));
  73     unixAddr.sun_family = AF_LOCAL;
  74     xstrncpy(unixAddr.sun_path, pathAddr.termedBuf(), sizeof(unixAddr.sun_path));
  75     return unixAddr;
  76 }
  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(fd(), 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 << "FD " << params.fd << " 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 fd() reopens?
 121     }
 122 }
 123
 124 void Ipc::UdsSender::timedout()
 125 {
 126     debugs(54, 5, HERE);
 127     mustStop("timedout");
 128 }
 129
 130
 131 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 132 {
 133     AsyncJob::Start(new UdsSender(toAddress, message));
 134 }
 135
 136 int Ipc::ImportFdIntoComm(int fd, int socktype, int protocol, Ipc::FdNoteId noteId)
 137 {
 138     struct sockaddr_in addr;
 139     socklen_t len = sizeof(addr);
 140     if (getsockname(fd, reinterpret_cast<sockaddr*>(&addr), &len) == 0) {
 141         Ip::Address ipAddr(addr);
 142         struct addrinfo* addr_info = NULL;
 143         ipAddr.GetAddrInfo(addr_info);
 144         addr_info->ai_socktype = socktype;
 145         addr_info->ai_protocol = protocol;
 146         comm_import_opened(fd, ipAddr, COMM_NONBLOCKING, Ipc::FdNote(noteId), addr_info);
 147         ipAddr.FreeAddrInfo(addr_info);
 148     } else {
 149         debugs(54, DBG_CRITICAL, HERE << "ERROR: FD " << fd << ' ' << xstrerror());
 150         ::close(fd);
 151         fd = -1;
 152     }
 153     return fd;
 154 }