src/control.cc

   1 #include <algorithm>
   2 #include <unordered_set>
   3
   4 #include "control.h"
   5 #include "service.h"
   6
   7 // Server-side control protocol implementation. This implements the functionality that allows
   8 // clients (such as dinitctl) to query service state and issue commands to control services.
   9
  10 namespace {
  11     constexpr auto OUT_EVENTS = dasynq::OUT_EVENTS;
  12     constexpr auto IN_EVENTS = dasynq::IN_EVENTS;
  13
  14     // Control protocol minimum compatible version and current version:
  15     constexpr uint16_t min_compat_version = 1;
  16     constexpr uint16_t cp_version = 1;
  17
  18     // check for value in a set
  19     template <typename T, int N>
  20     inline bool contains(const T (&v)[N], int i)
  21     {
  22         return std::find_if(std::begin(v), std::end(v),
  23                 [=](T p){ return i == static_cast<int>(p); }) != std::end(v);
  24     }
  25 }
  26
  27 bool control_conn_t::process_packet()
  28 {
  29     using std::string;
  30
  31     // Note that where we call queue_packet, we must generally check the return value. If it
  32     // returns false it has either deleted the connection or marked it for deletion; we
  33     // shouldn't touch instance members after that point.
  34
  35     int pktType = rbuf[0];
  36     if (pktType == DINIT_CP_QUERYVERSION) {
  37         // Responds with:
  38         // DINIT_RP_CVERSION, (2 byte) minimum compatible version, (2 byte) actual version
  39         char replyBuf[] = { DINIT_RP_CPVERSION, 0, 0, 0, 0 };
  40         memcpy(replyBuf + 1, &min_compat_version, 2);
  41         memcpy(replyBuf + 3, &cp_version, 2);
  42         if (! queue_packet(replyBuf, sizeof(replyBuf))) return false;
  43         rbuf.consume(1);
  44         return true;
  45     }
  46     if (pktType == DINIT_CP_FINDSERVICE || pktType == DINIT_CP_LOADSERVICE) {
  47         return process_find_load(pktType);
  48     }
  49     if (pktType == DINIT_CP_STARTSERVICE || pktType == DINIT_CP_STOPSERVICE
  50             || pktType == DINIT_CP_WAKESERVICE || pktType == DINIT_CP_RELEASESERVICE) {
  51         return process_start_stop(pktType);
  52     }
  53     if (pktType == DINIT_CP_UNPINSERVICE) {
  54         return process_unpin_service();
  55     }
  56     if (pktType == DINIT_CP_UNLOADSERVICE) {
  57         return process_unload_service();
  58     }
  59     if (pktType == DINIT_CP_SHUTDOWN) {
  60         // Shutdown/reboot
  61         if (rbuf.get_length() < 2) {
  62             chklen = 2;
  63             return true;
  64         }
  65
  66         if (contains({shutdown_type_t::CONTINUE, shutdown_type_t::HALT,
  67                 shutdown_type_t::POWEROFF, shutdown_type_t::REBOOT}, rbuf[1])) {
  68             auto sd_type = static_cast<shutdown_type_t>(rbuf[1]);
  69
  70             services->stop_all_services(sd_type);
  71             char ackBuf[] = { DINIT_RP_ACK };
  72             if (! queue_packet(ackBuf, 1)) return false;
  73
  74             // Clear the packet from the buffer
  75             rbuf.consume(2);
  76             chklen = 0;
  77             return true;
  78         }
  79
  80         // (otherwise fall through to below).
  81     }
  82     if (pktType == DINIT_CP_LISTSERVICES) {
  83         return list_services();
  84     }
  85     if (pktType == DINIT_CP_ADD_DEP) {
  86         return add_service_dep();
  87     }
  88     if (pktType == DINIT_CP_REM_DEP) {
  89         return rm_service_dep();
  90     }
  91     if (pktType == DINIT_CP_QUERY_LOAD_MECH) {
  92         return query_load_mech();
  93     }
  94
  95     // Unrecognized: give error response
  96     char outbuf[] = { DINIT_RP_BADREQ };
  97     if (! queue_packet(outbuf, 1)) return false;
  98     bad_conn_close = true;
  99     iob.set_watches(OUT_EVENTS);
 100     return true;
 101 }
 102
 103 bool control_conn_t::process_find_load(int pktType)
 104 {
 105     using std::string;
 106
 107     constexpr int pkt_size = 4;
 108
 109     if (rbuf.get_length() < pkt_size) {
 110         chklen = pkt_size;
 111         return true;
 112     }
 113
 114     uint16_t svcSize;
 115     rbuf.extract((char *)&svcSize, 1, 2);
 116     chklen = svcSize + 3; // packet type + (2 byte) length + service name
 117     if (svcSize <= 0 || chklen > 1024) {
 118         // Queue error response / mark connection bad
 119         char badreqRep[] = { DINIT_RP_BADREQ };
 120         if (! queue_packet(badreqRep, 1)) return false;
 121         bad_conn_close = true;
 122         iob.set_watches(OUT_EVENTS);
 123         return true;
 124     }
 125
 126     if (rbuf.get_length() < chklen) {
 127         // packet not complete yet; read more
 128         return true;
 129     }
 130
 131     service_record * record = nullptr;
 132
 133     string serviceName = rbuf.extract_string(3, svcSize);
 134
 135     if (pktType == DINIT_CP_LOADSERVICE) {
 136         // LOADSERVICE
 137         try {
 138             record = services->load_service(serviceName.c_str());
 139         }
 140         catch (service_load_exc &slexc) {
 141             log(loglevel_t::ERROR, "Could not load service ", slexc.serviceName, ": ", slexc.excDescription);
 142         }
 143     }
 144     else {
 145         // FINDSERVICE
 146         record = services->find_service(serviceName.c_str());
 147     }
 148
 149     if (record != nullptr) {
 150         // Allocate a service handle
 151         handle_t handle = allocate_service_handle(record);
 152         std::vector<char> rp_buf;
 153         rp_buf.reserve(7);
 154         rp_buf.push_back(DINIT_RP_SERVICERECORD);
 155         rp_buf.push_back(static_cast<char>(record->get_state()));
 156         for (int i = 0; i < (int) sizeof(handle); i++) {
 157             rp_buf.push_back(*(((char *) &handle) + i));
 158         }
 159         rp_buf.push_back(static_cast<char>(record->get_target_state()));
 160         if (! queue_packet(std::move(rp_buf))) return false;
 161     }
 162     else {
 163         std::vector<char> rp_buf = { DINIT_RP_NOSERVICE };
 164         if (! queue_packet(std::move(rp_buf))) return false;
 165     }
 166
 167     // Clear the packet from the buffer
 168     rbuf.consume(chklen);
 169     chklen = 0;
 170     return true;
 171 }
 172
 173 bool control_conn_t::process_start_stop(int pktType)
 174 {
 175     using std::string;
 176
 177     constexpr int pkt_size = 2 + sizeof(handle_t);
 178
 179     if (rbuf.get_length() < pkt_size) {
 180         chklen = pkt_size;
 181         return true;
 182     }
 183
 184     // 1 byte: packet type
 185     // 1 byte: pin in requested state (0 = no pin, 1 = pin)
 186     // 4 bytes: service handle
 187
 188     bool do_pin = (rbuf[1] == 1);
 189     handle_t handle;
 190     rbuf.extract((char *) &handle, 2, sizeof(handle));
 191
 192     service_record *service = find_service_for_key(handle);
 193     if (service == nullptr) {
 194         // Service handle is bad
 195         char badreqRep[] = { DINIT_RP_BADREQ };
 196         if (! queue_packet(badreqRep, 1)) return false;
 197         bad_conn_close = true;
 198         iob.set_watches(OUT_EVENTS);
 199         return true;
 200     }
 201     else {
 202         bool already_there = false;
 203
 204         switch (pktType) {
 205         case DINIT_CP_STARTSERVICE:
 206             // start service, mark as required
 207             if (do_pin) service->pin_start();
 208             service->start();
 209             services->process_queues();
 210             already_there = service->get_state() == service_state_t::STARTED;
 211             break;
 212         case DINIT_CP_STOPSERVICE:
 213             // force service to stop
 214             if (do_pin) service->pin_stop();
 215             service->stop(true);
 216             service->forced_stop();
 217             services->process_queues();
 218             already_there = service->get_state() == service_state_t::STOPPED;
 219             break;
 220         case DINIT_CP_WAKESERVICE:
 221             // re-start a stopped service (do not mark as required)
 222             if (do_pin) service->pin_start();
 223             service->start(false);
 224             services->process_queues();
 225             already_there = service->get_state() == service_state_t::STARTED;
 226             break;
 227         case DINIT_CP_RELEASESERVICE:
 228             // remove required mark, stop if not required by dependents
 229             if (do_pin) service->pin_stop();
 230             service->stop(false);
 231             services->process_queues();
 232             already_there = service->get_state() == service_state_t::STOPPED;
 233             break;
 234         }
 235
 236         char ack_buf[] = { (char)(already_there ? DINIT_RP_ALREADYSS : DINIT_RP_ACK) };
 237
 238         if (! queue_packet(ack_buf, 1)) return false;
 239     }
 240
 241     // Clear the packet from the buffer
 242     rbuf.consume(pkt_size);
 243     chklen = 0;
 244     return true;
 245 }
 246
 247 bool control_conn_t::process_unpin_service()
 248 {
 249     using std::string;
 250
 251     constexpr int pkt_size = 1 + sizeof(handle_t);
 252
 253     if (rbuf.get_length() < pkt_size) {
 254         chklen = pkt_size;
 255         return true;
 256     }
 257
 258     // 1 byte: packet type
 259     // 4 bytes: service handle
 260
 261     handle_t handle;
 262     rbuf.extract((char *) &handle, 1, sizeof(handle));
 263
 264     service_record *service = find_service_for_key(handle);
 265     if (service == nullptr) {
 266         // Service handle is bad
 267         char badreqRep[] = { DINIT_RP_BADREQ };
 268         if (! queue_packet(badreqRep, 1)) return false;
 269         bad_conn_close = true;
 270         iob.set_watches(OUT_EVENTS);
 271         return true;
 272     }
 273
 274     service->unpin();
 275     services->process_queues();
 276     char ack_buf[] = { (char) DINIT_RP_ACK };
 277     if (! queue_packet(ack_buf, 1)) return false;
 278
 279     // Clear the packet from the buffer
 280     rbuf.consume(pkt_size);
 281     chklen = 0;
 282     return true;
 283 }
 284
 285 bool control_conn_t::process_unload_service()
 286 {
 287     using std::string;
 288
 289     constexpr int pkt_size = 1 + sizeof(handle_t);
 290
 291     if (rbuf.get_length() < pkt_size) {
 292         chklen = pkt_size;
 293         return true;
 294     }
 295
 296     // 1 byte: packet type
 297     // 4 bytes: service handle
 298
 299     handle_t handle;
 300     rbuf.extract((char *) &handle, 1, sizeof(handle));
 301
 302     service_record *service = find_service_for_key(handle);
 303     if (service == nullptr) {
 304         // Service handle is bad
 305         char badreq_rep[] = { DINIT_RP_BADREQ };
 306         if (! queue_packet(badreq_rep, 1)) return false;
 307         bad_conn_close = true;
 308         iob.set_watches(OUT_EVENTS);
 309         return true;
 310     }
 311
 312     if (! service->has_lone_ref() || service->get_state() != service_state_t::STOPPED) {
 313         // Cannot unload: has other references
 314         char nak_rep[] = { DINIT_RP_NAK };
 315         if (! queue_packet(nak_rep, 1)) return false;
 316     }
 317     else {
 318         // unload
 319         service->prepare_for_unload();
 320         services->remove_service(service);
 321         delete service;
 322
 323         // drop handle
 324         service_key_map.erase(service);
 325         key_service_map.erase(handle);
 326
 327         // send ack
 328         char ack_buf[] = { (char) DINIT_RP_ACK };
 329         if (! queue_packet(ack_buf, 1)) return false;
 330     }
 331
 332     // Clear the packet from the buffer
 333     rbuf.consume(pkt_size);
 334     chklen = 0;
 335     return true;
 336 }
 337
 338 bool control_conn_t::list_services()
 339 {
 340     rbuf.consume(1); // clear request packet
 341     chklen = 0;
 342
 343     try {
 344         auto slist = services->list_services();
 345         for (auto sptr : slist) {
 346             std::vector<char> pkt_buf;
 347
 348             int hdrsize = 8 + std::max(sizeof(int), sizeof(pid_t));
 349
 350             const std::string &name = sptr->get_name();
 351             int nameLen = std::min((size_t)256, name.length());
 352             pkt_buf.resize(hdrsize + nameLen);
 353
 354             pkt_buf[0] = DINIT_RP_SVCINFO;
 355             pkt_buf[1] = nameLen;
 356             pkt_buf[2] = static_cast<char>(sptr->get_state());
 357             pkt_buf[3] = static_cast<char>(sptr->get_target_state());
 358
 359             char b0 = sptr->is_waiting_for_console() ? 1 : 0;
 360             b0 |= sptr->has_console() ? 2 : 0;
 361             b0 |= sptr->was_start_skipped() ? 4 : 0;
 362             pkt_buf[4] = b0;
 363             pkt_buf[5] = static_cast<char>(sptr->get_stop_reason());
 364
 365             pkt_buf[6] = 0; // reserved
 366             pkt_buf[7] = 0;
 367
 368             // Next: either the exit status, or the process ID
 369             if (sptr->get_state() != service_state_t::STOPPED) {
 370                 pid_t proc_pid = sptr->get_pid();
 371                 memcpy(pkt_buf.data() + 8, &proc_pid, sizeof(proc_pid));
 372             }
 373             else {
 374                 int exit_status = sptr->get_exit_status();
 375                 memcpy(pkt_buf.data() + 8, &exit_status, sizeof(exit_status));
 376             }
 377
 378             for (int i = 0; i < nameLen; i++) {
 379                 pkt_buf[hdrsize+i] = name[i];
 380             }
 381
 382             if (! queue_packet(std::move(pkt_buf))) return false;
 383         }
 384
 385         char ack_buf[] = { (char) DINIT_RP_LISTDONE };
 386         if (! queue_packet(ack_buf, 1)) return false;
 387
 388         return true;
 389     }
 390     catch (std::bad_alloc &exc)
 391     {
 392         do_oom_close();
 393         return true;
 394     }
 395 }
 396
 397 bool control_conn_t::add_service_dep()
 398 {
 399     // 1 byte packet type
 400     // 1 byte dependency type
 401     // handle: "from"
 402     // handle: "to"
 403
 404     constexpr int pkt_size = 2 + sizeof(handle_t) * 2;
 405
 406     if (rbuf.get_length() < pkt_size) {
 407         chklen = pkt_size;
 408         return true;
 409     }
 410
 411     handle_t from_handle;
 412     handle_t to_handle;
 413     rbuf.extract((char *) &from_handle, 2, sizeof(from_handle));
 414     rbuf.extract((char *) &to_handle, 2 + sizeof(from_handle), sizeof(to_handle));
 415
 416     service_record *from_service = find_service_for_key(from_handle);
 417     service_record *to_service = find_service_for_key(to_handle);
 418     if (from_service == nullptr || to_service == nullptr || from_service == to_service) {
 419         // Service handle is bad
 420         char badreq_rep[] = { DINIT_RP_BADREQ };
 421         if (! queue_packet(badreq_rep, 1)) return false;
 422         bad_conn_close = true;
 423         iob.set_watches(OUT_EVENTS);
 424         return true;
 425     }
 426
 427     // Check dependency type is valid:
 428     int dep_type_int = rbuf[1];
 429     if (! contains({(int)dependency_type::MILESTONE, (int)dependency_type::REGULAR,
 430             (int)dependency_type::WAITS_FOR}, dep_type_int)) {
 431         char badreqRep[] = { DINIT_RP_BADREQ };
 432         if (! queue_packet(badreqRep, 1)) return false;
 433         bad_conn_close = true;
 434         iob.set_watches(OUT_EVENTS);
 435     }
 436     dependency_type dep_type = static_cast<dependency_type>(dep_type_int);
 437
 438     // Check current service states are valid for given dep type
 439     if (dep_type == dependency_type::REGULAR) {
 440         if (from_service->get_state() != service_state_t::STOPPED &&
 441                 to_service->get_state() != service_state_t::STARTED) {
 442             // Cannot create dependency now since it would be contradicted:
 443             char nak_rep[] = { DINIT_RP_NAK };
 444             if (! queue_packet(nak_rep, 1)) return false;
 445             rbuf.consume(pkt_size);
 446             chklen = 0;
 447             return true;
 448         }
 449     }
 450
 451     // Check for creation of circular dependency chain
 452     std::unordered_set<service_record *> dep_marks;
 453     std::vector<service_record *> dep_queue;
 454     dep_queue.push_back(to_service);
 455     while (! dep_queue.empty()) {
 456         service_record * sr = dep_queue.back();
 457         dep_queue.pop_back();
 458         // iterate deps; if dep == from, abort; otherwise add to set/queue
 459         // (only add to queue if not already in set)
 460         for (auto &dep : sr->get_dependencies()) {
 461             service_record * dep_to = dep.get_to();
 462             if (dep_to == from_service) {
 463                 // fail, circular dependency!
 464                 char nak_rep[] = { DINIT_RP_NAK };
 465                 if (! queue_packet(nak_rep, 1)) return false;
 466                 rbuf.consume(pkt_size);
 467                 chklen = 0;
 468                 return true;
 469             }
 470             if (dep_marks.insert(dep_to).second) {
 471                 dep_queue.push_back(dep_to);
 472             }
 473         }
 474     }
 475     dep_marks.clear();
 476     dep_queue.clear();
 477
 478     // Prevent creation of duplicate dependency:
 479     for (auto &dep : from_service->get_dependencies()) {
 480         service_record * dep_to = dep.get_to();
 481         if (dep_to == to_service && dep.dep_type == dep_type) {
 482             // Dependency already exists: return success
 483             char ack_rep[] = { DINIT_RP_ACK };
 484             if (! queue_packet(ack_rep, 1)) return false;
 485             rbuf.consume(pkt_size);
 486             chklen = 0;
 487             return true;
 488         }
 489     }
 490
 491     // Create dependency:
 492     from_service->add_dep(to_service, dep_type);
 493     services->process_queues();
 494
 495     char ack_rep[] = { DINIT_RP_ACK };
 496     if (! queue_packet(ack_rep, 1)) return false;
 497     rbuf.consume(pkt_size);
 498     chklen = 0;
 499     return true;
 500 }
 501
 502 bool control_conn_t::rm_service_dep()
 503 {
 504     // 1 byte packet type
 505     // 1 byte dependency type
 506     // handle: "from"
 507     // handle: "to"
 508
 509     constexpr int pkt_size = 2 + sizeof(handle_t) * 2;
 510
 511     if (rbuf.get_length() < pkt_size) {
 512         chklen = pkt_size;
 513         return true;
 514     }
 515
 516     handle_t from_handle;
 517     handle_t to_handle;
 518     rbuf.extract((char *) &from_handle, 2, sizeof(from_handle));
 519     rbuf.extract((char *) &to_handle, 2 + sizeof(from_handle), sizeof(to_handle));
 520
 521     service_record *from_service = find_service_for_key(from_handle);
 522     service_record *to_service = find_service_for_key(to_handle);
 523     if (from_service == nullptr || to_service == nullptr || from_service == to_service) {
 524         // Service handle is bad
 525         char badreq_rep[] = { DINIT_RP_BADREQ };
 526         if (! queue_packet(badreq_rep, 1)) return false;
 527         bad_conn_close = true;
 528         iob.set_watches(OUT_EVENTS);
 529         return true;
 530     }
 531
 532     // Check dependency type is valid:
 533     int dep_type_int = rbuf[1];
 534     if (! contains({(int)dependency_type::MILESTONE, (int)dependency_type::REGULAR,
 535             (int)dependency_type::WAITS_FOR}, dep_type_int)) {
 536         char badreqRep[] = { DINIT_RP_BADREQ };
 537         if (! queue_packet(badreqRep, 1)) return false;
 538         bad_conn_close = true;
 539         iob.set_watches(OUT_EVENTS);
 540     }
 541     dependency_type dep_type = static_cast<dependency_type>(dep_type_int);
 542
 543     // Remove dependency:
 544     from_service->rm_dep(to_service, dep_type);
 545     services->process_queues();
 546
 547     char ack_rep[] = { DINIT_RP_ACK };
 548     if (! queue_packet(ack_rep, 1)) return false;
 549     rbuf.consume(pkt_size);
 550     chklen = 0;
 551     return true;
 552 }
 553
 554 bool control_conn_t::query_load_mech()
 555 {
 556     if (services->get_set_type_id() == SSET_TYPE_DIRLOAD) {
 557         dirload_service_set *dss = static_cast<dirload_service_set *>(services);
 558         std::vector<char> reppkt;
 559         reppkt.resize(2 + sizeof(uint32_t) * 2);  // packet type, loader type, packet size, # dirs
 560         reppkt[0] = DINIT_RP_LOADER_MECH;
 561         reppkt[1] = SSET_TYPE_DIRLOAD;
 562
 563         // Number of directories in load path:
 564         uint32_t sdirs = dss->get_service_dir_count();
 565         std::memcpy(reppkt.data() + 2 + sizeof(uint32_t), &sdirs, sizeof(sdirs));
 566
 567         for (int i = 0; i < sdirs; i++) {
 568             const char *sdir = dss->get_service_dir(i);
 569             uint32_t dlen = std::strlen(sdir);
 570             auto cursize = reppkt.size();
 571             reppkt.resize(cursize + sizeof(dlen) + dlen);
 572             std::memcpy(reppkt.data() + cursize, &dlen, sizeof(dlen));
 573             std::memcpy(reppkt.data() + cursize + sizeof(dlen), sdir, dlen);
 574         }
 575
 576         // Total packet size:
 577         uint32_t fsize = reppkt.size();
 578         std::memcpy(reppkt.data() + 2, &fsize, sizeof(fsize));
 579
 580         if (! queue_packet(std::move(reppkt))) return false;
 581         rbuf.consume(1);
 582         chklen = 0;
 583         return true;
 584     }
 585     else {
 586         // If we don't know how to deal with the service set type, send a NAK reply:
 587         char ack_rep[] = { DINIT_RP_NAK };
 588         if (! queue_packet(ack_rep, 1)) return false;
 589         rbuf.consume(1);
 590         chklen = 0;
 591         return true;
 592     }
 593 }
 594
 595 control_conn_t::handle_t control_conn_t::allocate_service_handle(service_record *record)
 596 {
 597     // Try to find a unique handle (integer) in a single pass. Since the map is ordered, we can search until
 598     // we find a gap in the handle values.
 599     handle_t candidate = 0;
 600     for (auto p : key_service_map) {
 601         if (p.first == candidate) candidate++;
 602         else break;
 603     }
 604
 605     bool is_unique = (service_key_map.find(record) == service_key_map.end());
 606
 607     // The following operations perform allocation (can throw std::bad_alloc). If an exception occurs we
 608     // must undo any previous actions:
 609     if (is_unique) {
 610         record->add_listener(this);
 611     }
 612
 613     try {
 614         key_service_map[candidate] = record;
 615         service_key_map.insert(std::make_pair(record, candidate));
 616     }
 617     catch (...) {
 618         if (is_unique) {
 619             record->remove_listener(this);
 620         }
 621
 622         key_service_map.erase(candidate);
 623     }
 624
 625     return candidate;
 626 }
 627
 628 bool control_conn_t::queue_packet(const char *pkt, unsigned size) noexcept
 629 {
 630     int in_flag = bad_conn_close ? 0 : IN_EVENTS;
 631     bool was_empty = outbuf.empty();
 632
 633     // If the queue is empty, we can try to write the packet out now rather than queueing it.
 634     // If the write is unsuccessful or partial, we queue the remainder.
 635     if (was_empty) {
 636         int wr = bp_sys::write(iob.get_watched_fd(), pkt, size);
 637         if (wr == -1) {
 638             if (errno == EPIPE) {
 639                 return false;
 640             }
 641             if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
 642                 log(loglevel_t::WARN, "Error writing to control connection: ", strerror(errno));
 643                 return false;
 644             }
 645             // EAGAIN etc: fall through to below
 646         }
 647         else {
 648             if ((unsigned)wr == size) {
 649                 // Ok, all written.
 650                 iob.set_watches(in_flag);
 651                 return true;
 652             }
 653             pkt += wr;
 654             size -= wr;
 655         }
 656     }
 657
 658     // Create a vector out of the (remaining part of the) packet:
 659     try {
 660         outbuf.emplace_back(pkt, pkt + size);
 661         iob.set_watches(in_flag | OUT_EVENTS);
 662         return true;
 663     }
 664     catch (std::bad_alloc &baexc) {
 665         // Mark the connection bad, and stop reading further requests
 666         bad_conn_close = true;
 667         oom_close = true;
 668         if (was_empty) {
 669             // We can't send out-of-memory response as we already wrote as much as we
 670             // could above. Neither can we later send the response since we have currently
 671             // sent an incomplete packet. All we can do is close the connection.
 672             return false;
 673         }
 674         else {
 675             iob.set_watches(OUT_EVENTS);
 676             return true;
 677         }
 678     }
 679 }
 680
 681 // This queue_packet method is frustratingly similar to the one above, but the subtle differences
 682 // make them extraordinary difficult to combine into a single method.
 683 bool control_conn_t::queue_packet(std::vector<char> &&pkt) noexcept
 684 {
 685     int in_flag = bad_conn_close ? 0 : IN_EVENTS;
 686     bool was_empty = outbuf.empty();
 687
 688     if (was_empty) {
 689         outpkt_index = 0;
 690         // We can try sending the packet immediately:
 691         int wr = bp_sys::write(iob.get_watched_fd(), pkt.data(), pkt.size());
 692         if (wr == -1) {
 693             if (errno == EPIPE) {
 694                 return false;
 695             }
 696             if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
 697                 log(loglevel_t::WARN, "Error writing to control connection: ", strerror(errno));
 698                 return false;
 699             }
 700             // EAGAIN etc: fall through to below
 701         }
 702         else {
 703             if ((unsigned)wr == pkt.size()) {
 704                 // Ok, all written.
 705                 iob.set_watches(in_flag);
 706                 return true;
 707             }
 708             outpkt_index = wr;
 709         }
 710     }
 711
 712     try {
 713         outbuf.emplace_back(pkt);
 714         iob.set_watches(in_flag | OUT_EVENTS);
 715         return true;
 716     }
 717     catch (std::bad_alloc &baexc) {
 718         // Mark the connection bad, and stop reading further requests
 719         bad_conn_close = true;
 720         oom_close = true;
 721         if (was_empty) {
 722             // We can't send out-of-memory response as we already wrote as much as we
 723             // could above. Neither can we later send the response since we have currently
 724             // sent an incomplete packet. All we can do is close the connection.
 725             return false;
 726         }
 727         else {
 728             iob.set_watches(OUT_EVENTS);
 729             return true;
 730         }
 731     }
 732 }
 733
 734 bool control_conn_t::rollback_complete() noexcept
 735 {
 736     char ackBuf[2] = { DINIT_ROLLBACK_COMPLETED, 2 };
 737     return queue_packet(ackBuf, 2);
 738 }
 739
 740 bool control_conn_t::data_ready() noexcept
 741 {
 742     int fd = iob.get_watched_fd();
 743
 744     int r = rbuf.fill(fd);
 745
 746     // Note file descriptor is non-blocking
 747     if (r == -1) {
 748         if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
 749             log(loglevel_t::WARN, "Error writing to control connection: ", strerror(errno));
 750             return true;
 751         }
 752         return false;
 753     }
 754
 755     if (r == 0) {
 756         return true;
 757     }
 758
 759     // complete packet?
 760     if (rbuf.get_length() >= chklen) {
 761         try {
 762             return !process_packet();
 763         }
 764         catch (std::bad_alloc &baexc) {
 765             do_oom_close();
 766             return false;
 767         }
 768     }
 769     else if (rbuf.get_length() == 1024) {
 770         // Too big packet
 771         log(loglevel_t::WARN, "Received too-large control package; dropping connection");
 772         bad_conn_close = true;
 773         iob.set_watches(OUT_EVENTS);
 774     }
 775     else {
 776         int out_flags = (bad_conn_close || !outbuf.empty()) ? OUT_EVENTS : 0;
 777         iob.set_watches(IN_EVENTS | out_flags);
 778     }
 779
 780     return false;
 781 }
 782
 783 bool control_conn_t::send_data() noexcept
 784 {
 785     if (outbuf.empty() && bad_conn_close) {
 786         if (oom_close) {
 787             // Send oom response
 788             char oomBuf[] = { DINIT_RP_OOM };
 789             bp_sys::write(iob.get_watched_fd(), oomBuf, 1);
 790         }
 791         return true;
 792     }
 793
 794     vector<char> & pkt = outbuf.front();
 795     char *data = pkt.data();
 796     int written = bp_sys::write(iob.get_watched_fd(), data + outpkt_index, pkt.size() - outpkt_index);
 797     if (written == -1) {
 798         if (errno == EPIPE) {
 799             // read end closed
 800             return true;
 801         }
 802         else if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
 803             // spurious readiness notification?
 804         }
 805         else {
 806             log(loglevel_t::ERROR, "Error writing to control connection: ", strerror(errno));
 807             return true;
 808         }
 809         return false;
 810     }
 811
 812     outpkt_index += written;
 813     if (outpkt_index == pkt.size()) {
 814         // We've finished this packet, move on to the next:
 815         outbuf.pop_front();
 816         outpkt_index = 0;
 817         if (outbuf.empty() && ! oom_close) {
 818             if (! bad_conn_close) {
 819                 iob.set_watches(IN_EVENTS);
 820             }
 821             else {
 822                 return true;
 823             }
 824         }
 825     }
 826
 827     return false;
 828 }
 829
 830 control_conn_t::~control_conn_t() noexcept
 831 {
 832     bp_sys::close(iob.get_watched_fd());
 833     iob.deregister(loop);
 834
 835     // Clear service listeners
 836     for (auto p : service_key_map) {
 837         p.first->remove_listener(this);
 838     }
 839
 840     active_control_conns--;
 841 }