src/control.cc

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