src/control.cc

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