src/ipc/UdsOp.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 54    Interprocess Communication
   5  *
   6  */
   7 #include "config.h"
   8 #include "base/TextException.h"
   9 #include "comm.h"
  10 #include "CommCalls.h"
  11 #include "comm/Connection.h"
  12 #include "ipc/UdsOp.h"
  13
  14
  15 Ipc::UdsOp::UdsOp(const String& pathAddr):
  16         AsyncJob("Ipc::UdsOp"),
  17         address(PathToAddress(pathAddr)),
  18         options(COMM_NONBLOCKING),
  19         fd_(-1)
  20 {
  21     debugs(54, 5, HERE << '[' << this << "] pathAddr=" << pathAddr);
  22 }
  23
  24 Ipc::UdsOp::~UdsOp()
  25 {
  26     debugs(54, 5, HERE << '[' << this << ']');
  27     if (fd_ >= 0)
  28         comm_close(fd_);
  29 }
  30
  31 void Ipc::UdsOp::setOptions(int newOptions)
  32 {
  33     options = newOptions;
  34 }
  35
  36 int Ipc::UdsOp::fd()
  37 {
  38     if (fd_ < 0) {
  39         if (options & COMM_DOBIND)
  40             unlink(address.sun_path);
  41         fd_ = comm_open_uds(SOCK_DGRAM, 0, &address, options);
  42         Must(fd_ >= 0);
  43     }
  44     return fd_;
  45 }
  46
  47 void Ipc::UdsOp::setTimeout(int seconds, const char *handlerName)
  48 {
  49     typedef CommCbMemFunT<UdsOp, CommTimeoutCbParams> Dialer;
  50     AsyncCall::Pointer handler = asyncCall(54,5, handlerName,
  51                                            Dialer(CbcPointer<UdsOp>(this), &UdsOp::noteTimeout));
  52     commSetTimeout(fd(), seconds, handler);
  53 }
  54
  55 void Ipc::UdsOp::clearTimeout()
  56 {
  57     commSetTimeout(fd(), -1, NULL, NULL); // TODO: add Comm::ClearTimeout(fd)
  58 }
  59
  60 void Ipc::UdsOp::noteTimeout(const CommTimeoutCbParams &)
  61 {
  62     timedout(); // our kid handles communication timeout
  63 }
  64
  65
  66 struct sockaddr_un
  67 Ipc::PathToAddress(const String& pathAddr) {
  68     assert(pathAddr.size() != 0);
  69     struct sockaddr_un unixAddr;
  70     memset(&unixAddr, 0, sizeof(unixAddr));
  71     unixAddr.sun_family = AF_LOCAL;
  72     xstrncpy(unixAddr.sun_path, pathAddr.termedBuf(), sizeof(unixAddr.sun_path));
  73     return unixAddr;
  74 }
  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(fd(), message.raw(), message.size(), writeHandler);
 109     writing = true;
 110 }
 111
 112 void Ipc::UdsSender::wrote(const CommIoCbParams& params)
 113 {
 114     debugs(54, 5, HERE << "FD " << params.fd << " flag " << params.flag << " [" << 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 fd() reopens?
 119     }
 120 }
 121
 122 void Ipc::UdsSender::timedout()
 123 {
 124     debugs(54, 5, HERE);
 125     mustStop("timedout");
 126 }
 127
 128
 129 void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message)
 130 {
 131     AsyncJob::Start(new UdsSender(toAddress, message));
 132 }