src/control.cc

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