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