control.cc

   1 #include "control.h"
   2 #include "service.h"
   3
   4 void ControlConn::processPacket()
   5 {
   6     using std::string;
   7
   8     // Note that where we call queuePacket, we must generally check the return value. If it
   9     // returns false it has either deleted the connection or marked it for deletion; we
  10     // shouldn't touch instance members after that point.
  11
  12     int pktType = rbuf[0];
  13     if (pktType == DINIT_CP_QUERYVERSION) {
  14         // Responds with:
  15         // DINIT_RP_CVERSION, (2 byte) minimum compatible version, (2 byte) maximum compatible version
  16         char replyBuf[] = { DINIT_RP_CPVERSION, 0, 0, 0, 0 };
  17         if (! queuePacket(replyBuf, 1)) return;
  18         rbuf.consume(1);
  19         return;
  20     }
  21     if (pktType == DINIT_CP_FINDSERVICE || pktType == DINIT_CP_LOADSERVICE) {
  22         processFindLoad(pktType);
  23         return;
  24     }
  25     if (pktType == DINIT_CP_STARTSERVICE || pktType == DINIT_CP_STOPSERVICE) {
  26         processStartStop(pktType);
  27         return;
  28     }
  29     else if (pktType == DINIT_CP_SHUTDOWN) {
  30         // Shutdown/reboot
  31         if (rbuf.get_length() < 2) {
  32             chklen = 2;
  33             return;
  34         }
  35
  36         auto sd_type = static_cast<ShutdownType>(rbuf[1]);
  37
  38         service_set->stop_all_services(sd_type);
  39         log_to_console = true;
  40         char ackBuf[] = { DINIT_RP_ACK };
  41         if (! queuePacket(ackBuf, 1)) return;
  42
  43         // Clear the packet from the buffer
  44         rbuf.consume(2);
  45         chklen = 0;
  46         return;
  47     }
  48     else {
  49         // Unrecognized: give error response
  50         char outbuf[] = { DINIT_RP_BADREQ };
  51         if (! queuePacket(outbuf, 1)) return;
  52         bad_conn_close = true;
  53         ev_io_set(&iob, iob.fd, EV_WRITE);
  54     }
  55     return;
  56 }
  57
  58 void ControlConn::processFindLoad(int pktType)
  59 {
  60     using std::string;
  61
  62     constexpr int pkt_size = 4;
  63
  64     if (rbuf.get_length() < pkt_size) {
  65         chklen = pkt_size;
  66         return;
  67     }
  68
  69     uint16_t svcSize;
  70     rbuf.extract((char *)&svcSize, 1, 2);
  71     chklen = svcSize + 3; // packet type + (2 byte) length + service name
  72     if (svcSize <= 0 || chklen > 1024) {
  73         // Queue error response / mark connection bad
  74         char badreqRep[] = { DINIT_RP_BADREQ };
  75         if (! queuePacket(badreqRep, 1)) return;
  76         bad_conn_close = true;
  77         ev_io_set(&iob, iob.fd, EV_WRITE);
  78         return;
  79     }
  80
  81     if (rbuf.get_length() < chklen) {
  82         // packet not complete yet; read more
  83         return;
  84     }
  85
  86     ServiceRecord * record = nullptr;
  87
  88     string serviceName = std::move(rbuf.extract_string(3, svcSize));
  89
  90     if (pktType == DINIT_CP_LOADSERVICE) {
  91         // LOADSERVICE
  92         try {
  93             record = service_set->loadService(serviceName);
  94         }
  95         catch (ServiceLoadExc &slexc) {
  96             log(LogLevel::ERROR, "Could not load service ", slexc.serviceName, ": ", slexc.excDescription);
  97         }
  98     }
  99     else {
 100         // FINDSERVICE
 101         record = service_set->findService(serviceName.c_str());
 102     }
 103
 104     if (record != nullptr) {
 105         // Allocate a service handle
 106         handle_t handle = allocateServiceHandle(record);
 107         std::vector<char> rp_buf;
 108         rp_buf.reserve(7);
 109         rp_buf.push_back(DINIT_RP_SERVICERECORD);
 110         rp_buf.push_back(static_cast<char>(record->getState()));
 111         for (int i = 0; i < (int) sizeof(handle); i++) {
 112             rp_buf.push_back(*(((char *) &handle) + i));
 113         }
 114         rp_buf.push_back(static_cast<char>(record->getTargetState()));
 115         if (! queuePacket(std::move(rp_buf))) return;
 116     }
 117     else {
 118         std::vector<char> rp_buf = { DINIT_RP_NOSERVICE };
 119         if (! queuePacket(std::move(rp_buf))) return;
 120     }
 121
 122     // Clear the packet from the buffer
 123     rbuf.consume(chklen);
 124     chklen = 0;
 125     return;
 126 }
 127
 128 void ControlConn::processStartStop(int pktType)
 129 {
 130     using std::string;
 131
 132     constexpr int pkt_size = 2 + sizeof(handle_t);
 133
 134     if (rbuf.get_length() < pkt_size) {
 135         chklen = pkt_size;
 136         return;
 137     }
 138
 139     // 1 byte: packet type
 140     // 1 byte: pin in requested state (0 = no pin, 1 = pin)
 141     // 4 bytes: service handle
 142
 143     bool do_pin = (rbuf[1] == 1);
 144     handle_t handle;
 145     rbuf.extract((char *) &handle, 2, sizeof(handle));
 146
 147     ServiceRecord *service = findServiceForKey(handle);
 148     if (service == nullptr) {
 149         // Service handle is bad
 150         char badreqRep[] = { DINIT_RP_BADREQ };
 151         if (! queuePacket(badreqRep, 1)) return;
 152         bad_conn_close = true;
 153         ev_io_set(&iob, iob.fd, EV_WRITE);
 154         return;
 155     }
 156     else {
 157         if (pktType == DINIT_CP_STARTSERVICE) {
 158             if (do_pin) {
 159                 service->pinStart();
 160             }
 161             else {
 162                 service->start();
 163             }
 164         }
 165         else {
 166             if (do_pin) {
 167                 service->pinStop();
 168             }
 169             else {
 170                 service->stop();
 171             }
 172         }
 173
 174         char ack_buf[] = { DINIT_RP_ACK };
 175         if (! queuePacket(ack_buf, 1)) return;
 176     }
 177
 178     // Clear the packet from the buffer
 179     rbuf.consume(pkt_size);
 180     chklen = 0;
 181     return;
 182 }
 183
 184 ControlConn::handle_t ControlConn::allocateServiceHandle(ServiceRecord *record)
 185 {
 186     bool is_unique = true;
 187     handle_t largest_seen = 0;
 188     handle_t candidate = 0;
 189     for (auto p : keyServiceMap) {
 190         if (p.first > largest_seen) largest_seen = p.first;
 191         if (p.first == candidate) {
 192             if (largest_seen == std::numeric_limits<handle_t>::max()) throw std::bad_alloc();
 193             candidate = largest_seen + 1;
 194         }
 195         is_unique &= (p.second != record);
 196     }
 197
 198     keyServiceMap[candidate] = record;
 199     serviceKeyMap.insert(std::make_pair(record, candidate));
 200
 201     if (is_unique) {
 202         record->addListener(this);
 203     }
 204
 205     return candidate;
 206 }
 207
 208
 209 bool ControlConn::queuePacket(const char *pkt, unsigned size) noexcept
 210 {
 211     if (bad_conn_close) return false;
 212
 213     bool was_empty = outbuf.empty();
 214
 215     if (was_empty) {
 216         int wr = write(iob.fd, pkt, size);
 217         if (wr == -1) {
 218             if (errno == EPIPE) {
 219                 delete this;
 220                 return false;
 221             }
 222             if (errno != EAGAIN && errno != EWOULDBLOCK) {
 223                 // TODO log error
 224                 delete this;
 225                 return false;
 226             }
 227         }
 228         else {
 229             if ((unsigned)wr == size) {
 230                 // Ok, all written.
 231                 return true;
 232             }
 233             pkt += wr;
 234             size -= wr;
 235         }
 236         ev_io_set(&iob, iob.fd, EV_READ | EV_WRITE);
 237     }
 238
 239     // Create a vector out of the (remaining part of the) packet:
 240     try {
 241         outbuf.emplace_back(pkt, pkt + size);
 242         return true;
 243     }
 244     catch (std::bad_alloc &baexc) {
 245         // Mark the connection bad, and stop reading further requests
 246         bad_conn_close = true;
 247         oom_close = true;
 248         if (was_empty) {
 249             // We can't send out-of-memory response as we already wrote as much as we
 250             // could above. Neither can we later send the response since we have currently
 251             // sent an incomplete packet. All we can do is close the connection.
 252             delete this;
 253         }
 254         else {
 255             ev_io_set(&iob, iob.fd, EV_WRITE);
 256         }
 257         return false;
 258     }
 259 }
 260
 261
 262 bool ControlConn::queuePacket(std::vector<char> &&pkt) noexcept
 263 {
 264     if (bad_conn_close) return false;
 265
 266     bool was_empty = outbuf.empty();
 267
 268     if (was_empty) {
 269         outpkt_index = 0;
 270         // We can try sending the packet immediately:
 271         int wr = write(iob.fd, pkt.data(), pkt.size());
 272         if (wr == -1) {
 273             if (errno == EPIPE) {
 274                 delete this;
 275                 return false;
 276             }
 277             if (errno != EAGAIN && errno != EWOULDBLOCK) {
 278                 // TODO log error
 279                 delete this;
 280                 return false;
 281             }
 282         }
 283         else {
 284             if ((unsigned)wr == pkt.size()) {
 285                 // Ok, all written.
 286                 return true;
 287             }
 288             outpkt_index = wr;
 289         }
 290         ev_io_set(&iob, iob.fd, EV_READ | EV_WRITE);
 291     }
 292
 293     try {
 294         outbuf.emplace_back(pkt);
 295         return true;
 296     }
 297     catch (std::bad_alloc &baexc) {
 298         // Mark the connection bad, and stop reading further requests
 299         bad_conn_close = true;
 300         oom_close = true;
 301         if (was_empty) {
 302             // We can't send out-of-memory response as we already wrote as much as we
 303             // could above. Neither can we later send the response since we have currently
 304             // sent an incomplete packet. All we can do is close the connection.
 305             delete this;
 306         }
 307         else {
 308             ev_io_set(&iob, iob.fd, EV_WRITE);
 309         }
 310         return false;
 311     }
 312 }
 313
 314 bool ControlConn::rollbackComplete() noexcept
 315 {
 316     char ackBuf[2] = { DINIT_ROLLBACK_COMPLETED, 2 };
 317     return queuePacket(ackBuf, 2);
 318 }
 319
 320 bool ControlConn::dataReady() noexcept
 321 {
 322     int fd = iob.fd;
 323
 324     int r = rbuf.fill(fd);
 325
 326     // Note file descriptor is non-blocking
 327     if (r == -1) {
 328         if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
 329             // TODO log error
 330             delete this;
 331             return true;
 332         }
 333         return false;
 334     }
 335
 336     if (r == 0) {
 337         delete this;
 338         return true;
 339     }
 340
 341     // complete packet?
 342     if (rbuf.get_length() >= chklen) {
 343         try {
 344             processPacket();
 345         }
 346         catch (std::bad_alloc &baexc) {
 347             doOomClose();
 348         }
 349     }
 350
 351     if (rbuf.get_length() == 1024) {
 352         // Too big packet
 353         // TODO log error?
 354         // TODO error response?
 355         bad_conn_close = true;
 356         ev_io_set(&iob, iob.fd, EV_WRITE);
 357     }
 358
 359     return false;
 360 }
 361
 362 void ControlConn::sendData() noexcept
 363 {
 364     if (outbuf.empty() && bad_conn_close) {
 365         if (oom_close) {
 366             // Send oom response
 367             char oomBuf[] = { DINIT_RP_OOM };
 368             write(iob.fd, oomBuf, 1);
 369         }
 370         delete this;
 371         return;
 372     }
 373
 374     vector<char> & pkt = outbuf.front();
 375     char *data = pkt.data();
 376     int written = write(iob.fd, data + outpkt_index, pkt.size() - outpkt_index);
 377     if (written == -1) {
 378         if (errno == EPIPE) {
 379             // read end closed
 380             delete this;
 381         }
 382         else if (errno == EAGAIN || errno == EWOULDBLOCK) {
 383             // spurious readiness notification?
 384         }
 385         else {
 386             log(LogLevel::ERROR, "Error writing to control connection: ", strerror(errno));
 387             delete this;
 388         }
 389         return;
 390     }
 391
 392     outpkt_index += written;
 393     if (outpkt_index == pkt.size()) {
 394         // We've finished this packet, move on to the next:
 395         outbuf.pop_front();
 396         outpkt_index = 0;
 397         if (outbuf.empty() && ! oom_close) {
 398             if (! bad_conn_close) {
 399                 ev_io_set(&iob, iob.fd, EV_READ);
 400             }
 401             else {
 402                 delete this;
 403             }
 404         }
 405     }
 406 }
 407
 408 ControlConn::~ControlConn() noexcept
 409 {
 410     close(iob.fd);
 411     ev_io_stop(loop, &iob);
 412
 413     // Clear service listeners
 414     for (auto p : serviceKeyMap) {
 415         p.first->removeListener(this);
 416     }
 417
 418     active_control_conns--;
 419 }