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 "base/TextException.h"
  13 #include "ipc/UdsOp.h"
  14
  15
  16 Ipc::UdsOp::UdsOp(const String& pathAddr):
  17         AsyncJob("Ipc::UdsOp"),
  18         address(PathToAddress(pathAddr)),
  19         options(COMM_NONBLOCKING),
  20         fd_(-1)
  21 {
  22     debugs(54, 5, HERE << '[' << this << "] pathAddr=" << pathAddr);
  23 }
  24
  25 Ipc::UdsOp::~UdsOp()
  26 {
  27     debugs(54, 5, HERE << '[' << this << ']');
  28     if (fd_ >= 0)
  29         comm_close(fd_);
  30 }
  31
  32 void Ipc::UdsOp::setOptions(int newOptions)
  33 {
  34     options = newOptions;
  35 }
  36
  37 int Ipc::UdsOp::fd()
  38 {
  39     if (fd_ < 0) {
  40         if (options & COMM_DOBIND)
  41             unlink(address.sun_path);
  42         fd_ = comm_open_uds(SOCK_DGRAM, 0, &address, options);
  43         Must(fd_ >= 0);
  44     }
  45     return fd_;
  46 }
  47
  48 void Ipc::UdsOp::setTimeout(int seconds, const char *handlerName)
  49 {
  50     typedef CommCbMemFunT<UdsOp, CommTimeoutCbParams> Dialer;
  51     AsyncCall::Pointer handler = asyncCall(54,5, handlerName,
  52                                            Dialer(CbcPointer<UdsOp>(this), &UdsOp::noteTimeout));
  53     commSetTimeout(fd(), seconds, handler);
  54 }
  55
  56 void Ipc::UdsOp::clearTimeout()
  57 {
  58     commSetTimeout(fd(), -1, NULL, NULL); // TODO: add Comm::ClearTimeout(fd)
  59 }
  60
  61 void Ipc::UdsOp::noteTimeout(const CommTimeoutCbParams &)
  62 {
  63     timedout(); // our kid handles communication timeout
  64 }
  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
  78 CBDATA_NAMESPACED_CLASS_INIT(Ipc, UdsSender);
  79
  80 Ipc::UdsSender::UdsSender(const String& pathAddr, const TypedMsgHdr& aMessage):
  81         UdsOp(pathAddr),
  82         message(aMessage),
  83         retries(10), // TODO: make configurable?
  84         timeout(10), // TODO: make configurable?
  85         writing(false)
  86 {
  87     message.address(address);
  88 }
  89
  90 void Ipc::UdsSender::start()
  91 {
  92     UdsOp::start();
  93     write();
  94     if (timeout > 0)
  95         setTimeout(timeout, "Ipc::UdsSender::noteTimeout");
  96 }
  97
  98 bool Ipc::UdsSender::doneAll() const
  99 {
 100     return !writing && UdsOp::doneAll();
 101 }
 102
 103 void Ipc::UdsSender::write()
 104 {
 105     debugs(54, 5, HERE);
 106     typedef CommCbMemFunT<UdsSender, CommIoCbParams> Dialer;
 107     AsyncCall::Pointer writeHandler = JobCallback(54, 5,
 108                                                    Dialer, this, UdsSender::wrote);
 109     comm_write(fd(), message.raw(), message.size(), writeHandler);
 110     writing = true;
 111 }
 112
 113 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 114 {
 115     debugs(54, 5, HERE << "FD " << params.fd << " flag " << params.flag << " [" << this << ']');
 116     writing = false;
 117     if (params.flag != COMM_OK && retries-- > 0) {
 118         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?
 119         write(); // XXX: should we close on error so that fd() reopens?
 120     }
 121 }
 122
 123 void Ipc::UdsSender::timedout()
 124 {
 125     debugs(54, 5, HERE);
 126     mustStop("timedout");
 127 }
 128
 129
 130 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 131 {
 132     AsyncJob::Start(new UdsSender(toAddress, message));
 133 }