control.cc 43 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377
  1. #include <algorithm>
  2. #include <unordered_set>
  3. #include <climits>
  4. #include "dinit-env.h"
  5. #include "control.h"
  6. #include "service.h"
  7. #include "proc-service.h"
  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. // Control protocol versions:
  11. // 1 - dinit 0.16 and prior
  12. // 2 - dinit 0.17 (adds DINIT_CP_SETTRIGGER, DINIT_CP_CATLOG, DINIT_CP_SIGNAL)
  13. namespace {
  14. // Control protocol minimum compatible version and current version:
  15. constexpr uint16_t min_compat_version = 1;
  16. constexpr uint16_t cp_version = 2;
  17. // check for value in a set
  18. template <typename T, int N, typename U>
  19. inline bool contains(const T (&v)[N], U i)
  20. {
  21. return std::find_if(std::begin(v), std::end(v),
  22. [=](T p){ return i == static_cast<U>(p); }) != std::end(v);
  23. }
  24. }
  25. bool control_conn_t::process_packet()
  26. {
  27. using std::string;
  28. // Note that where we call queue_packet, we must generally check the return value. If it
  29. // returns false it has either deleted the connection or marked it for deletion; we
  30. // shouldn't touch instance members after that point.
  31. int pktType = rbuf[0];
  32. if (pktType == DINIT_CP_QUERYVERSION) {
  33. // Responds with:
  34. // DINIT_RP_CVERSION, (2 byte) minimum compatible version, (2 byte) actual version
  35. char replyBuf[] = { DINIT_RP_CPVERSION, 0, 0, 0, 0 };
  36. memcpy(replyBuf + 1, &min_compat_version, 2);
  37. memcpy(replyBuf + 3, &cp_version, 2);
  38. if (!queue_packet(replyBuf, sizeof(replyBuf))) return false;
  39. rbuf.consume(1);
  40. return true;
  41. }
  42. if (pktType == DINIT_CP_FINDSERVICE || pktType == DINIT_CP_LOADSERVICE) {
  43. return process_find_load(pktType);
  44. }
  45. if (pktType == DINIT_CP_STARTSERVICE || pktType == DINIT_CP_STOPSERVICE
  46. || pktType == DINIT_CP_WAKESERVICE || pktType == DINIT_CP_RELEASESERVICE) {
  47. return process_start_stop(pktType);
  48. }
  49. if (pktType == DINIT_CP_UNPINSERVICE) {
  50. return process_unpin_service();
  51. }
  52. if (pktType == DINIT_CP_UNLOADSERVICE) {
  53. return process_unload_service();
  54. }
  55. if (pktType == DINIT_CP_RELOADSERVICE) {
  56. return process_reload_service();
  57. }
  58. if (pktType == DINIT_CP_SHUTDOWN) {
  59. // Shutdown/reboot
  60. if (rbuf.get_length() < 2) {
  61. chklen = 2;
  62. return true;
  63. }
  64. if (contains({shutdown_type_t::REMAIN, shutdown_type_t::HALT,
  65. shutdown_type_t::POWEROFF, shutdown_type_t::REBOOT}, rbuf[1])) {
  66. auto sd_type = static_cast<shutdown_type_t>(rbuf[1]);
  67. services->stop_all_services(sd_type);
  68. char ackBuf[] = { DINIT_RP_ACK };
  69. if (! queue_packet(ackBuf, 1)) return false;
  70. // Clear the packet from the buffer
  71. rbuf.consume(2);
  72. chklen = 0;
  73. return true;
  74. }
  75. // (otherwise fall through to below).
  76. }
  77. if (pktType == DINIT_CP_LISTSERVICES) {
  78. return list_services();
  79. }
  80. if (pktType == DINIT_CP_SERVICESTATUS) {
  81. return process_service_status();
  82. }
  83. if (pktType == DINIT_CP_ADD_DEP) {
  84. return add_service_dep();
  85. }
  86. if (pktType == DINIT_CP_REM_DEP) {
  87. return rm_service_dep();
  88. }
  89. if (pktType == DINIT_CP_QUERY_LOAD_MECH) {
  90. return query_load_mech();
  91. }
  92. if (pktType == DINIT_CP_ENABLESERVICE) {
  93. return add_service_dep(true);
  94. }
  95. if (pktType == DINIT_CP_QUERYSERVICENAME) {
  96. return process_query_name();
  97. }
  98. if (pktType == DINIT_CP_SETENV) {
  99. return process_setenv();
  100. }
  101. if (pktType == DINIT_CP_SETTRIGGER) {
  102. return process_set_trigger();
  103. }
  104. if (pktType == DINIT_CP_CATLOG) {
  105. return process_catlog();
  106. }
  107. if (pktType == DINIT_CP_SIGNAL) {
  108. return process_signal();
  109. }
  110. // Unrecognized: give error response
  111. char outbuf[] = { DINIT_RP_BADREQ };
  112. if (!queue_packet(outbuf, 1)) return false;
  113. bad_conn_close = true;
  114. return true;
  115. }
  116. bool control_conn_t::process_find_load(int pktType)
  117. {
  118. using std::string;
  119. constexpr int pkt_size = 4;
  120. if (rbuf.get_length() < pkt_size) {
  121. chklen = pkt_size;
  122. return true;
  123. }
  124. uint16_t svcSize;
  125. rbuf.extract((char *)&svcSize, 1, 2);
  126. if (svcSize <= 0 || svcSize > (1024 - 3)) {
  127. // Queue error response / mark connection bad
  128. char badreqRep[] = { DINIT_RP_BADREQ };
  129. if (! queue_packet(badreqRep, 1)) return false;
  130. bad_conn_close = true;
  131. return true;
  132. }
  133. chklen = svcSize + 3; // packet type + (2 byte) length + service name
  134. if (rbuf.get_length() < chklen) {
  135. // packet not complete yet; read more
  136. return true;
  137. }
  138. service_record * record = nullptr;
  139. string serviceName = rbuf.extract_string(3, svcSize);
  140. // Clear the packet from the buffer
  141. rbuf.consume(chklen);
  142. chklen = 0;
  143. char fail_code = DINIT_RP_NOSERVICE;
  144. if (pktType == DINIT_CP_LOADSERVICE) {
  145. // LOADSERVICE
  146. try {
  147. record = services->load_service(serviceName.c_str());
  148. }
  149. catch (service_description_exc &sdexc) {
  150. log_service_load_failure(sdexc);
  151. fail_code = DINIT_RP_SERVICE_DESC_ERR;
  152. }
  153. catch (service_not_found &snf) {
  154. log(loglevel_t::ERROR, "Could not load service ", snf.service_name, ": ",
  155. snf.exc_description);
  156. // fail_code = DINIT_RP_NOSERVICE; (already set)
  157. }
  158. catch (service_load_exc &slexc) {
  159. log(loglevel_t::ERROR, "Could not load service ", slexc.service_name, ": ",
  160. slexc.exc_description);
  161. fail_code = DINIT_RP_SERVICE_LOAD_ERR;
  162. }
  163. }
  164. else {
  165. // FINDSERVICE
  166. record = services->find_service(serviceName.c_str());
  167. }
  168. if (record == nullptr) {
  169. std::vector<char> rp_buf = { fail_code };
  170. if (! queue_packet(std::move(rp_buf))) return false;
  171. return true;
  172. }
  173. // Allocate a service handle
  174. handle_t handle = allocate_service_handle(record);
  175. std::vector<char> rp_buf;
  176. rp_buf.reserve(7);
  177. rp_buf.push_back(DINIT_RP_SERVICERECORD);
  178. rp_buf.push_back(static_cast<char>(record->get_state()));
  179. for (int i = 0; i < (int) sizeof(handle); i++) {
  180. rp_buf.push_back(*(((char *) &handle) + i));
  181. }
  182. rp_buf.push_back(static_cast<char>(record->get_target_state()));
  183. if (! queue_packet(std::move(rp_buf))) return false;
  184. return true;
  185. }
  186. bool control_conn_t::check_dependents(service_record *service, bool &had_dependents)
  187. {
  188. std::vector<char> reply_pkt;
  189. size_t num_depts = 0;
  190. for (service_dep *dep : service->get_dependents()) {
  191. if (dep->dep_type == dependency_type::REGULAR && dep->holding_acq) {
  192. num_depts++;
  193. // find or allocate a service handle
  194. handle_t dept_handle = allocate_service_handle(dep->get_from());
  195. if (reply_pkt.empty()) {
  196. // packet type, size
  197. reply_pkt.reserve(1 + sizeof(size_t) + sizeof(handle_t));
  198. reply_pkt.resize(1 + sizeof(size_t));
  199. reply_pkt[0] = DINIT_RP_DEPENDENTS;
  200. }
  201. auto old_size = reply_pkt.size();
  202. reply_pkt.resize(old_size + sizeof(handle_t));
  203. memcpy(reply_pkt.data() + old_size, &dept_handle, sizeof(dept_handle));
  204. }
  205. }
  206. if (num_depts != 0) {
  207. // There are affected dependents
  208. had_dependents = true;
  209. memcpy(reply_pkt.data() + 1, &num_depts, sizeof(num_depts));
  210. return queue_packet(std::move(reply_pkt));
  211. }
  212. had_dependents = false;
  213. return true;
  214. }
  215. bool control_conn_t::process_start_stop(int pktType)
  216. {
  217. using std::string;
  218. constexpr int pkt_size = 2 + sizeof(handle_t);
  219. if (rbuf.get_length() < pkt_size) {
  220. chklen = pkt_size;
  221. return true;
  222. }
  223. // 1 byte: packet type
  224. // 1 byte: flags eg. pin in requested state (0 = no pin, 1 = pin)
  225. // 4 bytes: service handle
  226. bool do_pin = ((rbuf[1] & 1) == 1);
  227. handle_t handle;
  228. rbuf.extract((char *) &handle, 2, sizeof(handle));
  229. service_record *service = find_service_for_key(handle);
  230. if (service == nullptr) {
  231. // Service handle is bad
  232. char badreqRep[] = { DINIT_RP_BADREQ };
  233. if (! queue_packet(badreqRep, 1)) return false;
  234. bad_conn_close = true;
  235. return true;
  236. }
  237. else {
  238. char ack_buf[1] = { DINIT_RP_ACK };
  239. switch (pktType) {
  240. case DINIT_CP_STARTSERVICE:
  241. // start service, mark as required
  242. if (services->is_shutting_down()) {
  243. ack_buf[0] = DINIT_RP_SHUTTINGDOWN;
  244. break;
  245. }
  246. if ((service->get_state() == service_state_t::STOPPED
  247. || service->get_state() == service_state_t::STOPPING)
  248. && service->is_stop_pinned()) {
  249. ack_buf[0] = DINIT_RP_PINNEDSTOPPED;
  250. break;
  251. }
  252. if (do_pin) service->pin_start();
  253. service->start();
  254. services->process_queues();
  255. if (service->get_state() == service_state_t::STARTED) ack_buf[0] = DINIT_RP_ALREADYSS;
  256. break;
  257. case DINIT_CP_STOPSERVICE:
  258. {
  259. // force service to stop
  260. bool do_restart = ((rbuf[1] & 4) == 4);
  261. bool gentle = ((rbuf[1] & 2) == 2);
  262. if (do_restart && services->is_shutting_down()) {
  263. ack_buf[0] = DINIT_RP_SHUTTINGDOWN;
  264. break;
  265. }
  266. if ((service->get_state() == service_state_t::STARTED
  267. || service->get_state() == service_state_t::STARTING)
  268. && service->is_start_pinned()) {
  269. ack_buf[0] = DINIT_RP_PINNEDSTARTED;
  270. break;
  271. }
  272. if (gentle) {
  273. // Check dependents; return appropriate response if any will be affected
  274. bool has_dependents;
  275. if (!check_dependents(service, has_dependents)) {
  276. return false;
  277. }
  278. if (has_dependents) {
  279. // Reply packet has already been sent
  280. goto clear_out;
  281. }
  282. }
  283. service_state_t wanted_state;
  284. if (do_restart) {
  285. if (! service->restart()) {
  286. ack_buf[0] = DINIT_RP_NAK;
  287. break;
  288. }
  289. wanted_state = service_state_t::STARTED;
  290. }
  291. else {
  292. if (do_pin) service->pin_stop();
  293. service->stop(true);
  294. service->forced_stop();
  295. wanted_state = service_state_t::STOPPED;
  296. }
  297. services->process_queues();
  298. if (service->get_state() == wanted_state && !do_restart) ack_buf[0] = DINIT_RP_ALREADYSS;
  299. break;
  300. }
  301. case DINIT_CP_WAKESERVICE:
  302. {
  303. // re-attach a service to its (started) dependents, causing it to start.
  304. if (services->is_shutting_down()) {
  305. ack_buf[0] = DINIT_RP_SHUTTINGDOWN;
  306. break;
  307. }
  308. if ((service->get_state() == service_state_t::STOPPED
  309. || service->get_state() == service_state_t::STOPPING)
  310. && service->is_stop_pinned()) {
  311. ack_buf[0] = DINIT_RP_PINNEDSTOPPED;
  312. break;
  313. }
  314. bool found_dpt = false;
  315. for (auto dpt : service->get_dependents()) {
  316. if (dpt->is_only_ordering()) continue;
  317. auto from = dpt->get_from();
  318. auto from_state = from->get_state();
  319. if (from_state == service_state_t::STARTED || from_state == service_state_t::STARTING) {
  320. found_dpt = true;
  321. if (!dpt->holding_acq) {
  322. dpt->get_from()->start_dep(*dpt);
  323. }
  324. }
  325. }
  326. if (!found_dpt) {
  327. ack_buf[0] = DINIT_RP_NAK;
  328. }
  329. if (do_pin) service->pin_start();
  330. services->process_queues();
  331. if (service->get_state() == service_state_t::STARTED) ack_buf[0] = DINIT_RP_ALREADYSS;
  332. break;
  333. }
  334. case DINIT_CP_RELEASESERVICE:
  335. // remove required mark, stop if not required by dependents
  336. if (do_pin) service->pin_stop();
  337. service->stop(false);
  338. services->process_queues();
  339. if (service->get_state() == service_state_t::STOPPED) ack_buf[0] = DINIT_RP_ALREADYSS;
  340. break;
  341. }
  342. if (! queue_packet(ack_buf, 1)) return false;
  343. }
  344. clear_out:
  345. // Clear the packet from the buffer
  346. rbuf.consume(pkt_size);
  347. chklen = 0;
  348. return true;
  349. }
  350. bool control_conn_t::process_unpin_service()
  351. {
  352. using std::string;
  353. constexpr int pkt_size = 1 + sizeof(handle_t);
  354. if (rbuf.get_length() < pkt_size) {
  355. chklen = pkt_size;
  356. return true;
  357. }
  358. // 1 byte: packet type
  359. // 4 bytes: service handle
  360. handle_t handle;
  361. rbuf.extract((char *) &handle, 1, sizeof(handle));
  362. service_record *service = find_service_for_key(handle);
  363. if (service == nullptr) {
  364. // Service handle is bad
  365. char badreqRep[] = { DINIT_RP_BADREQ };
  366. if (! queue_packet(badreqRep, 1)) return false;
  367. bad_conn_close = true;
  368. return true;
  369. }
  370. service->unpin();
  371. services->process_queues();
  372. char ack_buf[] = { (char) DINIT_RP_ACK };
  373. if (! queue_packet(ack_buf, 1)) return false;
  374. // Clear the packet from the buffer
  375. rbuf.consume(pkt_size);
  376. chklen = 0;
  377. return true;
  378. }
  379. bool control_conn_t::process_unload_service()
  380. {
  381. using std::string;
  382. constexpr int pkt_size = 1 + sizeof(handle_t);
  383. if (rbuf.get_length() < pkt_size) {
  384. chklen = pkt_size;
  385. return true;
  386. }
  387. // 1 byte: packet type
  388. // 4 bytes: service handle
  389. handle_t handle;
  390. rbuf.extract((char *) &handle, 1, sizeof(handle));
  391. service_record *service = find_service_for_key(handle);
  392. if (service == nullptr) {
  393. // Service handle is bad
  394. char badreq_rep[] = { DINIT_RP_BADREQ };
  395. if (! queue_packet(badreq_rep, 1)) return false;
  396. bad_conn_close = true;
  397. return true;
  398. }
  399. if (!service->has_lone_ref() || service->get_state() != service_state_t::STOPPED) {
  400. // Cannot unload: has other references
  401. char nak_rep[] = { DINIT_RP_NAK };
  402. if (!queue_packet(nak_rep, 1)) return false;
  403. }
  404. else {
  405. // unload (this may fail with bad_alloc)
  406. services->unload_service(service);
  407. // drop handle
  408. service_key_map.erase(service);
  409. key_service_map.erase(handle);
  410. // send ack
  411. char ack_buf[] = { (char) DINIT_RP_ACK };
  412. if (!queue_packet(ack_buf, 1)) return false;
  413. }
  414. // Clear the packet from the buffer
  415. rbuf.consume(pkt_size);
  416. chklen = 0;
  417. return true;
  418. }
  419. bool control_conn_t::process_reload_service()
  420. {
  421. using std::string;
  422. constexpr int pkt_size = 1 + sizeof(handle_t);
  423. if (rbuf.get_length() < pkt_size) {
  424. chklen = pkt_size;
  425. return true;
  426. }
  427. // 1 byte: packet type
  428. // 4 bytes: service handle
  429. handle_t handle;
  430. rbuf.extract((char *) &handle, 1, sizeof(handle));
  431. service_record *service = find_service_for_key(handle);
  432. if (service == nullptr) {
  433. // Service handle is bad
  434. char badreq_rep[] = { DINIT_RP_BADREQ };
  435. if (! queue_packet(badreq_rep, 1)) return false;
  436. bad_conn_close = true;
  437. return true;
  438. }
  439. if (!service->has_lone_ref(false)) {
  440. // Cannot unload: has other references
  441. char nak_rep[] = { DINIT_RP_NAK };
  442. if (! queue_packet(nak_rep, 1)) return false;
  443. }
  444. else {
  445. try {
  446. // drop handle
  447. key_service_map.erase(handle);
  448. service_key_map.erase(service);
  449. // reload
  450. service->remove_listener(this);
  451. services->reload_service(service);
  452. services->process_queues();
  453. // send ack
  454. char ack_buf[] = { (char) DINIT_RP_ACK };
  455. if (! queue_packet(ack_buf, 1)) return false;
  456. }
  457. catch (service_load_exc &slexc) {
  458. log(loglevel_t::ERROR, "Could not reload service ", slexc.service_name, ": ",
  459. slexc.exc_description);
  460. char nak_rep[] = { DINIT_RP_NAK };
  461. if (! queue_packet(nak_rep, 1)) return false;
  462. }
  463. }
  464. // Clear the packet from the buffer
  465. rbuf.consume(pkt_size);
  466. chklen = 0;
  467. return true;
  468. }
  469. constexpr static unsigned SIZEOF_INT_PIDT_UNION = ((sizeof(pid_t) > sizeof(int)) ? sizeof(pid_t) : sizeof(int));
  470. constexpr static unsigned STATUS_BUFFER_SIZE = 6 + SIZEOF_INT_PIDT_UNION;
  471. static void fill_status_buffer(char *buffer, service_record *service)
  472. {
  473. buffer[0] = static_cast<char>(service->get_state());
  474. buffer[1] = static_cast<char>(service->get_target_state());
  475. pid_t proc_pid = service->get_pid();
  476. char b0 = service->is_waiting_for_console() ? 1 : 0;
  477. b0 |= service->has_console() ? 2 : 0;
  478. b0 |= service->was_start_skipped() ? 4 : 0;
  479. b0 |= service->is_marked_active() ? 8 : 0;
  480. b0 |= (proc_pid != -1) ? 16 : 0;
  481. buffer[2] = b0;
  482. buffer[3] = static_cast<char>(service->get_stop_reason());
  483. buffer[4] = 0; // (if exec failed, these are replaced with stage)
  484. buffer[5] = 0;
  485. if (proc_pid != -1) {
  486. memcpy(buffer + 6, &proc_pid, sizeof(proc_pid));
  487. }
  488. else {
  489. // These values only make sense in STOPPING/STOPPED, but we'll fill them in regardless:
  490. if (buffer[3] == (char)stopped_reason_t::EXECFAILED) {
  491. base_process_service *bsp = (base_process_service *)service;
  492. run_proc_err exec_err = bsp->get_exec_err_info();
  493. uint16_t stage = (uint16_t)exec_err.stage;
  494. memcpy(buffer + 4, &stage, 2);
  495. memcpy(buffer + 6, &exec_err.st_errno, sizeof(int));
  496. }
  497. else {
  498. int exit_status = service->get_exit_status();
  499. memcpy(buffer + 6, &exit_status, sizeof(exit_status));
  500. }
  501. }
  502. }
  503. bool control_conn_t::list_services()
  504. {
  505. rbuf.consume(1); // clear request packet
  506. chklen = 0;
  507. try {
  508. auto slist = services->list_services();
  509. for (auto sptr : slist) {
  510. if (sptr->get_type() == service_type_t::PLACEHOLDER) continue;
  511. std::vector<char> pkt_buf;
  512. int hdrsize = 2 + STATUS_BUFFER_SIZE;
  513. const std::string &name = sptr->get_name();
  514. int nameLen = std::min((size_t)256, name.length());
  515. pkt_buf.resize(hdrsize + nameLen);
  516. pkt_buf[0] = DINIT_RP_SVCINFO;
  517. pkt_buf[1] = nameLen;
  518. fill_status_buffer(&pkt_buf[2], sptr);
  519. for (int i = 0; i < nameLen; i++) {
  520. pkt_buf[hdrsize+i] = name[i];
  521. }
  522. if (!queue_packet(std::move(pkt_buf))) return false;
  523. }
  524. char ack_buf[] = { (char) DINIT_RP_LISTDONE };
  525. if (! queue_packet(ack_buf, 1)) return false;
  526. return true;
  527. }
  528. catch (std::bad_alloc &exc)
  529. {
  530. do_oom_close();
  531. return true;
  532. }
  533. }
  534. bool control_conn_t::process_service_status()
  535. {
  536. constexpr int pkt_size = 1 + sizeof(handle_t);
  537. if (rbuf.get_length() < pkt_size) {
  538. chklen = pkt_size;
  539. return true;
  540. }
  541. handle_t handle;
  542. rbuf.extract(&handle, 1, sizeof(handle));
  543. rbuf.consume(pkt_size);
  544. chklen = 0;
  545. service_record *service = find_service_for_key(handle);
  546. if (service == nullptr || service->get_name().length() > std::numeric_limits<uint16_t>::max()) {
  547. char nak_rep[] = { DINIT_RP_NAK };
  548. return queue_packet(nak_rep, 1);
  549. }
  550. // Reply:
  551. // 1 byte packet type = DINIT_RP_SERVICESTATUS
  552. // 1 byte reserved ( = 0)
  553. // STATUS_BUFFER_SIZE bytes status
  554. std::vector<char> pkt_buf(2 + STATUS_BUFFER_SIZE);
  555. pkt_buf[0] = DINIT_RP_SERVICESTATUS;
  556. pkt_buf[1] = 0;
  557. fill_status_buffer(pkt_buf.data() + 2, service);
  558. return queue_packet(std::move(pkt_buf));
  559. }
  560. bool control_conn_t::add_service_dep(bool do_enable)
  561. {
  562. // 1 byte packet type
  563. // 1 byte dependency type
  564. // handle: "from"
  565. // handle: "to"
  566. constexpr int pkt_size = 2 + sizeof(handle_t) * 2;
  567. if (rbuf.get_length() < pkt_size) {
  568. chklen = pkt_size;
  569. return true;
  570. }
  571. handle_t from_handle;
  572. handle_t to_handle;
  573. rbuf.extract((char *) &from_handle, 2, sizeof(from_handle));
  574. rbuf.extract((char *) &to_handle, 2 + sizeof(from_handle), sizeof(to_handle));
  575. service_record *from_service = find_service_for_key(from_handle);
  576. service_record *to_service = find_service_for_key(to_handle);
  577. if (from_service == nullptr || to_service == nullptr || from_service == to_service) {
  578. // Service handle is bad
  579. char badreq_rep[] = { DINIT_RP_BADREQ };
  580. if (!queue_packet(badreq_rep, 1)) return false;
  581. bad_conn_close = true;
  582. return true;
  583. }
  584. // Check dependency type is valid:
  585. int dep_type_int = rbuf[1];
  586. if (!contains({dependency_type::MILESTONE, dependency_type::REGULAR,
  587. dependency_type::WAITS_FOR}, dep_type_int)) {
  588. char badreqRep[] = { DINIT_RP_BADREQ };
  589. if (!queue_packet(badreqRep, 1)) return false;
  590. bad_conn_close = true;
  591. }
  592. dependency_type dep_type = static_cast<dependency_type>(dep_type_int);
  593. // Check current service states are valid for given dep type
  594. if (dep_type == dependency_type::REGULAR) {
  595. if (from_service->get_state() != service_state_t::STOPPED &&
  596. to_service->get_state() != service_state_t::STARTED) {
  597. // Cannot create dependency now since it would be contradicted:
  598. char nak_rep[] = { DINIT_RP_NAK };
  599. if (! queue_packet(nak_rep, 1)) return false;
  600. rbuf.consume(pkt_size);
  601. chklen = 0;
  602. return true;
  603. }
  604. }
  605. // Check for creation of circular dependency chain
  606. std::unordered_set<service_record *> dep_marks;
  607. std::vector<service_record *> dep_queue;
  608. dep_queue.push_back(to_service);
  609. while (! dep_queue.empty()) {
  610. service_record * sr = dep_queue.back();
  611. dep_queue.pop_back();
  612. // iterate deps; if dep == from, abort; otherwise add to set/queue
  613. // (only add to queue if not already in set)
  614. for (auto &dep : sr->get_dependencies()) {
  615. service_record * dep_to = dep.get_to();
  616. if (dep_to == from_service) {
  617. // fail, circular dependency!
  618. char nak_rep[] = { DINIT_RP_NAK };
  619. if (! queue_packet(nak_rep, 1)) return false;
  620. rbuf.consume(pkt_size);
  621. chklen = 0;
  622. return true;
  623. }
  624. if (dep_marks.insert(dep_to).second) {
  625. dep_queue.push_back(dep_to);
  626. }
  627. }
  628. }
  629. dep_marks.clear();
  630. dep_queue.clear();
  631. bool dep_exists = false;
  632. service_dep * dep_record = nullptr;
  633. // Prevent creation of duplicate dependency:
  634. for (auto &dep : from_service->get_dependencies()) {
  635. service_record * dep_to = dep.get_to();
  636. if (dep_to == to_service && dep.dep_type == dep_type) {
  637. // Dependency already exists
  638. dep_exists = true;
  639. dep_record = &dep;
  640. break;
  641. }
  642. }
  643. if (! dep_exists) {
  644. // Create dependency:
  645. dep_record = &(from_service->add_dep(to_service, dep_type));
  646. services->process_queues();
  647. }
  648. if (do_enable && contains({service_state_t::STARTED, service_state_t::STARTING},
  649. from_service->get_state())) {
  650. // The dependency record is activated: mark it as holding acquisition of the dependency, and start
  651. // the dependency.
  652. if (!services->is_shutting_down()) {
  653. dep_record->get_from()->start_dep(*dep_record);
  654. services->process_queues();
  655. }
  656. }
  657. char ack_rep[] = { DINIT_RP_ACK };
  658. if (! queue_packet(ack_rep, 1)) return false;
  659. rbuf.consume(pkt_size);
  660. chklen = 0;
  661. return true;
  662. }
  663. bool control_conn_t::rm_service_dep()
  664. {
  665. // 1 byte packet type
  666. // 1 byte dependency type
  667. // handle: "from"
  668. // handle: "to"
  669. constexpr int pkt_size = 2 + sizeof(handle_t) * 2;
  670. if (rbuf.get_length() < pkt_size) {
  671. chklen = pkt_size;
  672. return true;
  673. }
  674. handle_t from_handle;
  675. handle_t to_handle;
  676. rbuf.extract((char *) &from_handle, 2, sizeof(from_handle));
  677. rbuf.extract((char *) &to_handle, 2 + sizeof(from_handle), sizeof(to_handle));
  678. service_record *from_service = find_service_for_key(from_handle);
  679. service_record *to_service = find_service_for_key(to_handle);
  680. if (from_service == nullptr || to_service == nullptr || from_service == to_service) {
  681. // Service handle is bad
  682. char badreq_rep[] = { DINIT_RP_BADREQ };
  683. if (! queue_packet(badreq_rep, 1)) return false;
  684. bad_conn_close = true;
  685. return true;
  686. }
  687. // Check dependency type is valid:
  688. int dep_type_int = rbuf[1];
  689. if (! contains({dependency_type::MILESTONE, dependency_type::REGULAR,
  690. dependency_type::WAITS_FOR}, dep_type_int)) {
  691. char badreqRep[] = { DINIT_RP_BADREQ };
  692. if (! queue_packet(badreqRep, 1)) return false;
  693. bad_conn_close = true;
  694. }
  695. dependency_type dep_type = static_cast<dependency_type>(dep_type_int);
  696. // Remove dependency:
  697. bool did_remove = from_service->rm_dep(to_service, dep_type);
  698. services->process_queues();
  699. char ack_rep[] = { did_remove ? (char)DINIT_RP_ACK : (char)DINIT_RP_NAK };
  700. if (! queue_packet(ack_rep, 1)) return false;
  701. rbuf.consume(pkt_size);
  702. chklen = 0;
  703. return true;
  704. }
  705. bool control_conn_t::process_query_name()
  706. {
  707. // 1 byte packet type
  708. // 1 byte reserved
  709. // handle: service
  710. constexpr int pkt_size = 2 + sizeof(handle_t);
  711. if (rbuf.get_length() < pkt_size) {
  712. chklen = pkt_size;
  713. return true;
  714. }
  715. handle_t handle;
  716. rbuf.extract(&handle, 2, sizeof(handle));
  717. rbuf.consume(pkt_size);
  718. chklen = 0;
  719. service_record *service = find_service_for_key(handle);
  720. if (service == nullptr || service->get_name().length() > std::numeric_limits<uint16_t>::max()) {
  721. char nak_rep[] = { DINIT_RP_NAK };
  722. return queue_packet(nak_rep, 1);
  723. }
  724. // Reply:
  725. // 1 byte packet type = DINIT_RP_SERVICENAME
  726. // 1 byte reserved
  727. // uint16_t length
  728. // N bytes name
  729. std::vector<char> reply;
  730. const std::string &name = service->get_name();
  731. uint16_t name_length = name.length();
  732. reply.resize(2 + sizeof(uint16_t) + name_length);
  733. reply[0] = DINIT_RP_SERVICENAME;
  734. memcpy(reply.data() + 2, &name_length, sizeof(name_length));
  735. memcpy(reply.data() + 2 + sizeof(uint16_t), name.c_str(), name_length);
  736. return queue_packet(std::move(reply));
  737. }
  738. bool control_conn_t::process_setenv()
  739. {
  740. using std::string;
  741. string envVar;
  742. typename string::size_type eq;
  743. constexpr int pkt_size = 4;
  744. char badreqRep[] = { DINIT_RP_BADREQ };
  745. char okRep[] = { DINIT_RP_ACK };
  746. if (rbuf.get_length() < pkt_size) {
  747. chklen = pkt_size;
  748. return true;
  749. }
  750. uint16_t envSize;
  751. rbuf.extract((char *)&envSize, 1, 2);
  752. if (envSize <= 0 || envSize > (1024 - 3)) {
  753. goto badreq;
  754. }
  755. chklen = envSize + 3; // packet type + (2 byte) length + envvar
  756. if (rbuf.get_length() < chklen) {
  757. // packet not complete yet; read more
  758. return true;
  759. }
  760. envVar = rbuf.extract_string(3, envSize);
  761. eq = envVar.find('=');
  762. if (!eq || eq == envVar.npos) {
  763. // Not found or at the beginning of the string
  764. goto badreq;
  765. }
  766. main_env.set_var(std::move(envVar));
  767. // Success response
  768. if (!queue_packet(okRep, 1)) return false;
  769. // Clear the packet from the buffer
  770. rbuf.consume(chklen);
  771. chklen = 0;
  772. return true;
  773. badreq:
  774. // Queue error response / mark connection bad
  775. if (! queue_packet(badreqRep, 1)) return false;
  776. bad_conn_close = true;
  777. return true;
  778. }
  779. bool control_conn_t::process_set_trigger()
  780. {
  781. // 1 byte packet type
  782. // handle: service
  783. // 1 byte trigger value
  784. constexpr int pkt_size = 2 + sizeof(handle_t);
  785. if (rbuf.get_length() < pkt_size) {
  786. chklen = pkt_size;
  787. return true;
  788. }
  789. handle_t handle;
  790. char trigger_val;
  791. rbuf.extract(&handle, 1, sizeof(handle));
  792. rbuf.extract(&trigger_val, 1 + sizeof(handle), sizeof(trigger_val));
  793. rbuf.consume(pkt_size);
  794. chklen = 0;
  795. service_record *service = find_service_for_key(handle);
  796. if (service == nullptr || service->get_type() != service_type_t::TRIGGERED) {
  797. char nak_rep[] = { DINIT_RP_NAK };
  798. return queue_packet(nak_rep, 1);
  799. }
  800. triggered_service *tservice = static_cast<triggered_service *>(service);
  801. tservice->set_trigger(trigger_val != 0);
  802. services->process_queues();
  803. char ack_rep[] = { DINIT_RP_ACK };
  804. return queue_packet(ack_rep, 1);
  805. }
  806. bool control_conn_t::process_catlog()
  807. {
  808. // 1 byte packet type
  809. // 1 byte reserved for future use
  810. // handle
  811. constexpr int pkt_size = 2 + sizeof(handle_t);
  812. if (rbuf.get_length() < pkt_size) {
  813. chklen = pkt_size;
  814. return true;
  815. }
  816. handle_t handle;
  817. char flags = rbuf[1];
  818. rbuf.extract(&handle, 2, sizeof(handle));
  819. rbuf.consume(pkt_size);
  820. chklen = 0;
  821. service_record *service = find_service_for_key(handle);
  822. if (service == nullptr || (service->get_type() != service_type_t::PROCESS
  823. && service->get_type() != service_type_t::BGPROCESS
  824. && service->get_type() != service_type_t::SCRIPTED)) {
  825. char nak_rep[] = { DINIT_RP_NAK };
  826. return queue_packet(nak_rep, 1);
  827. }
  828. base_process_service *bps = static_cast<base_process_service *>(service);
  829. if (bps->get_log_mode() != log_type_id::BUFFER) {
  830. char nak_rep[] = { DINIT_RP_NAK };
  831. return queue_packet(nak_rep, 1);
  832. }
  833. auto buffer_details = bps->get_log_buffer();
  834. const char *bufaddr = buffer_details.first;
  835. unsigned buflen = buffer_details.second;
  836. std::vector<char> pkt = { (char)DINIT_RP_SERVICE_LOG, 0 /* flags; reserved for future */ };
  837. pkt.insert(pkt.end(), (char *)(&buflen), (char *)(&buflen + 1));
  838. pkt.insert(pkt.end(), bufaddr, bufaddr + buflen);
  839. if ((flags & 1) != 0) {
  840. bps->clear_log_buffer();
  841. }
  842. return queue_packet(std::move(pkt));
  843. }
  844. bool control_conn_t::process_signal()
  845. {
  846. // packet contains signal number and process handle
  847. constexpr int pkt_size = 1 + sizeof(int) + sizeof(handle_t);
  848. if (rbuf.get_length() < pkt_size) {
  849. chklen = pkt_size;
  850. return true;
  851. }
  852. int sig_num;
  853. rbuf.extract((char *) &sig_num, 1, sizeof(sig_num));
  854. handle_t handle;
  855. rbuf.extract((char *) &handle, 1 + sizeof(sig_num), sizeof(handle));
  856. rbuf.consume(pkt_size);
  857. chklen = 0;
  858. service_record *service = find_service_for_key(handle);
  859. if (service == nullptr) {
  860. char nak_rep[] = { DINIT_RP_NAK };
  861. return queue_packet(nak_rep, 1);
  862. }
  863. // Reply:
  864. // 1 byte packet type = DINIT_RP_*
  865. pid_t spid = service->get_pid();
  866. // we probably don't want to kill/signal every process (in the current group),
  867. // but get_pid() sometimes returns -1 if e.g. service is not 'started'
  868. if (spid == -1 || spid == 0) {
  869. char nak_rep[] = { DINIT_RP_SIGNAL_NOPID };
  870. return queue_packet(nak_rep, 1);
  871. }
  872. else {
  873. if (bp_sys::kill(spid, sig_num) != 0) {
  874. if (errno == EINVAL) {
  875. log(loglevel_t::ERROR, "Requested signal not in valid signal range.");
  876. char nak_rep[] = { DINIT_RP_SIGNAL_BADSIG };
  877. return queue_packet(nak_rep, 1);
  878. }
  879. log(loglevel_t::ERROR, "Error sending signal to process: ", strerror(errno));
  880. char nak_rep[] = { DINIT_RP_SIGNAL_KILLERR };
  881. return queue_packet(nak_rep, 1);
  882. }
  883. }
  884. char ack_rep[] = { DINIT_RP_ACK };
  885. return queue_packet(ack_rep, 1);
  886. }
  887. bool control_conn_t::query_load_mech()
  888. {
  889. rbuf.consume(1);
  890. chklen = 0;
  891. if (services->get_set_type_id() == SSET_TYPE_DIRLOAD) {
  892. dirload_service_set *dss = static_cast<dirload_service_set *>(services);
  893. std::vector<char> reppkt;
  894. reppkt.resize(2 + sizeof(uint32_t) * 2); // packet type, loader type, packet size, # dirs
  895. reppkt[0] = DINIT_RP_LOADER_MECH;
  896. reppkt[1] = SSET_TYPE_DIRLOAD;
  897. // Number of directories in load path:
  898. uint32_t sdirs = dss->get_service_dir_count();
  899. std::memcpy(reppkt.data() + 2 + sizeof(uint32_t), &sdirs, sizeof(sdirs));
  900. // Our current working directory, which above are relative to:
  901. // leave sizeof(uint32_t) for size, which we'll fill in afterwards:
  902. std::size_t curpos = reppkt.size() + sizeof(uint32_t);
  903. #ifdef PATH_MAX
  904. uint32_t try_path_size = PATH_MAX;
  905. #else
  906. uint32_t try_path_size = 2048;
  907. #endif
  908. char *wd;
  909. while (true) {
  910. std::size_t total_size = curpos + std::size_t(try_path_size);
  911. if (total_size < curpos) {
  912. // Overflow. In theory we could now limit to size_t max, but the size must already
  913. // be crazy long; let's abort.
  914. char ack_rep[] = { DINIT_RP_NAK };
  915. if (! queue_packet(ack_rep, 1)) return false;
  916. return true;
  917. }
  918. reppkt.resize(total_size);
  919. wd = getcwd(reppkt.data() + curpos, try_path_size);
  920. if (wd != nullptr) break;
  921. // Keep doubling the path size we try until it's big enough, or we get numeric overflow
  922. uint32_t new_try_path_size = try_path_size * uint32_t(2u);
  923. if (new_try_path_size < try_path_size) {
  924. // Overflow.
  925. char ack_rep[] = { DINIT_RP_NAK };
  926. return queue_packet(ack_rep, 1);
  927. }
  928. try_path_size = new_try_path_size;
  929. }
  930. uint32_t wd_len = std::strlen(reppkt.data() + curpos);
  931. reppkt.resize(curpos + std::size_t(wd_len));
  932. std::memcpy(reppkt.data() + curpos - sizeof(uint32_t), &wd_len, sizeof(wd_len));
  933. // Each directory in the load path:
  934. for (int i = 0; uint32_t(i) < sdirs; i++) {
  935. const char *sdir = dss->get_service_dir(i);
  936. uint32_t dlen = std::strlen(sdir);
  937. auto cursize = reppkt.size();
  938. reppkt.resize(cursize + sizeof(dlen) + dlen);
  939. std::memcpy(reppkt.data() + cursize, &dlen, sizeof(dlen));
  940. std::memcpy(reppkt.data() + cursize + sizeof(dlen), sdir, dlen);
  941. }
  942. // Total packet size:
  943. uint32_t fsize = reppkt.size();
  944. std::memcpy(reppkt.data() + 2, &fsize, sizeof(fsize));
  945. if (! queue_packet(std::move(reppkt))) return false;
  946. return true;
  947. }
  948. else {
  949. // If we don't know how to deal with the service set type, send a NAK reply:
  950. char ack_rep[] = { DINIT_RP_NAK };
  951. return queue_packet(ack_rep, 1);
  952. }
  953. }
  954. control_conn_t::handle_t control_conn_t::allocate_service_handle(service_record *record)
  955. {
  956. // Try to find a unique handle (integer) in a single pass. Since the map is ordered, we can search until
  957. // we find a gap in the handle values.
  958. handle_t candidate = 0;
  959. for (auto p : key_service_map) {
  960. if (p.first == candidate) ++candidate;
  961. else break;
  962. }
  963. bool is_unique = (service_key_map.find(record) == service_key_map.end());
  964. // The following operations perform allocation (can throw std::bad_alloc). If an exception occurs we
  965. // must undo any previous actions:
  966. if (is_unique) {
  967. record->add_listener(this);
  968. }
  969. try {
  970. key_service_map[candidate] = record;
  971. service_key_map.insert(std::make_pair(record, candidate));
  972. }
  973. catch (...) {
  974. if (is_unique) {
  975. record->remove_listener(this);
  976. }
  977. key_service_map.erase(candidate);
  978. }
  979. return candidate;
  980. }
  981. void control_conn_t::service_event(service_record *service, service_event_t event) noexcept
  982. {
  983. // For each service handle corresponding to the event, send an information packet.
  984. auto range = service_key_map.equal_range(service);
  985. auto &i = range.first;
  986. auto &end = range.second;
  987. try {
  988. while (i != end) {
  989. uint32_t key = i->second;
  990. std::vector<char> pkt;
  991. constexpr int pktsize = 3 + sizeof(key) + STATUS_BUFFER_SIZE;
  992. pkt.reserve(pktsize);
  993. pkt.push_back(DINIT_IP_SERVICEEVENT);
  994. pkt.push_back(pktsize);
  995. char *p = (char *)&key;
  996. for (unsigned j = 0; j < sizeof(key); j++) {
  997. pkt.push_back(*p++);
  998. }
  999. pkt.push_back(static_cast<char>(event));
  1000. pkt.resize(pktsize);
  1001. fill_status_buffer(pkt.data() + 3 + sizeof(key), service);
  1002. queue_packet(std::move(pkt));
  1003. ++i;
  1004. }
  1005. }
  1006. catch (std::bad_alloc &exc) {
  1007. do_oom_close();
  1008. }
  1009. }
  1010. bool control_conn_t::queue_packet(const char *pkt, unsigned size) noexcept
  1011. {
  1012. bool was_empty = outbuf.empty();
  1013. // If the queue is empty, we can try to write the packet out now rather than queueing it.
  1014. // If the write is unsuccessful or partial, we queue the remainder.
  1015. if (was_empty) {
  1016. int wr = bp_sys::write(iob.get_watched_fd(), pkt, size);
  1017. if (wr == -1) {
  1018. if (errno == EPIPE) {
  1019. return false;
  1020. }
  1021. if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
  1022. log(loglevel_t::WARN, "Error writing to control connection: ", strerror(errno));
  1023. return false;
  1024. }
  1025. // EAGAIN etc: fall through to below
  1026. }
  1027. else {
  1028. if ((unsigned)wr == size) {
  1029. // Ok, all written.
  1030. return true;
  1031. }
  1032. pkt += wr;
  1033. size -= wr;
  1034. }
  1035. }
  1036. // Create a vector out of the (remaining part of the) packet:
  1037. try {
  1038. outbuf.emplace_back(pkt, pkt + size);
  1039. outbuf_size += size;
  1040. return true;
  1041. }
  1042. catch (std::bad_alloc &baexc) {
  1043. // Mark the connection bad, and stop reading further requests
  1044. bad_conn_close = true;
  1045. oom_close = true;
  1046. if (was_empty) {
  1047. // We can't send out-of-memory response as we already wrote as much as we
  1048. // could above. Neither can we later send the response since we have currently
  1049. // sent an incomplete packet. All we can do is close the connection.
  1050. return false;
  1051. }
  1052. else {
  1053. return true;
  1054. }
  1055. }
  1056. }
  1057. // This queue_packet method is frustratingly similar to the one above, but the subtle differences
  1058. // make them extraordinary difficult to combine into a single method.
  1059. bool control_conn_t::queue_packet(std::vector<char> &&pkt) noexcept
  1060. {
  1061. bool was_empty = outbuf.empty();
  1062. if (was_empty) {
  1063. outpkt_index = 0;
  1064. // We can try sending the packet immediately:
  1065. int wr = bp_sys::write(iob.get_watched_fd(), pkt.data(), pkt.size());
  1066. if (wr == -1) {
  1067. if (errno == EPIPE) {
  1068. return false;
  1069. }
  1070. if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
  1071. log(loglevel_t::WARN, "Error writing to control connection: ", strerror(errno));
  1072. return false;
  1073. }
  1074. // EAGAIN etc: fall through to below
  1075. }
  1076. else {
  1077. if ((unsigned)wr == pkt.size()) {
  1078. // Ok, all written.
  1079. return true;
  1080. }
  1081. outpkt_index = wr;
  1082. }
  1083. }
  1084. try {
  1085. outbuf.emplace_back(std::move(pkt));
  1086. outbuf_size += pkt.size();
  1087. return true;
  1088. }
  1089. catch (std::bad_alloc &baexc) {
  1090. // Mark the connection bad, and stop reading further requests
  1091. bad_conn_close = true;
  1092. oom_close = true;
  1093. if (was_empty) {
  1094. // We can't send out-of-memory response as we already wrote as much as we
  1095. // could above. Neither can we later send the response since we have currently
  1096. // sent an incomplete packet. All we can do is close the connection.
  1097. return false;
  1098. }
  1099. else {
  1100. return true;
  1101. }
  1102. }
  1103. }
  1104. bool control_conn_t::data_ready() noexcept
  1105. {
  1106. int fd = iob.get_watched_fd();
  1107. int r = rbuf.fill(fd);
  1108. // Note file descriptor is non-blocking
  1109. if (r == -1) {
  1110. if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
  1111. if (errno != ECONNRESET) {
  1112. log(loglevel_t::WARN, "Error reading from control connection: ", strerror(errno));
  1113. }
  1114. return true;
  1115. }
  1116. return false;
  1117. }
  1118. if (r == 0) {
  1119. return true;
  1120. }
  1121. // complete packet?
  1122. if (rbuf.get_length() >= chklen) {
  1123. try {
  1124. return !process_packet();
  1125. }
  1126. catch (std::bad_alloc &baexc) {
  1127. do_oom_close();
  1128. return false;
  1129. }
  1130. }
  1131. else if (rbuf.get_length() == rbuf.get_size()) {
  1132. // Too big packet
  1133. log(loglevel_t::WARN, "Received too-large control packet; dropping connection");
  1134. bad_conn_close = true;
  1135. }
  1136. return false;
  1137. }
  1138. bool control_conn_t::send_data() noexcept
  1139. {
  1140. if (outbuf.empty() && bad_conn_close) {
  1141. if (oom_close) {
  1142. // Send oom response
  1143. char oomBuf[] = { DINIT_RP_OOM };
  1144. bp_sys::write(iob.get_watched_fd(), oomBuf, 1);
  1145. }
  1146. return true;
  1147. }
  1148. vector<char> & pkt = outbuf.front();
  1149. char *data = pkt.data();
  1150. int written = bp_sys::write(iob.get_watched_fd(), data + outpkt_index, pkt.size() - outpkt_index);
  1151. if (written == -1) {
  1152. if (errno == EPIPE) {
  1153. // read end closed
  1154. return true;
  1155. }
  1156. else if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
  1157. // spurious readiness notification?
  1158. return false;
  1159. }
  1160. else {
  1161. log(loglevel_t::ERROR, "Error writing to control connection: ", strerror(errno));
  1162. return true;
  1163. }
  1164. }
  1165. outpkt_index += written;
  1166. if (outpkt_index == pkt.size()) {
  1167. // We've finished this packet, move on to the next:
  1168. outbuf_size -= pkt.size();
  1169. outbuf.pop_front();
  1170. outpkt_index = 0;
  1171. if (oom_close) {
  1172. // remain active, try to send DINIT_RP_OOM shortly
  1173. return false;
  1174. }
  1175. if (outbuf.empty() && bad_conn_close) {
  1176. return true;
  1177. }
  1178. }
  1179. // more to send
  1180. return false;
  1181. }
  1182. control_conn_t::~control_conn_t() noexcept
  1183. {
  1184. int fd = iob.get_watched_fd();
  1185. iob.deregister(loop);
  1186. bp_sys::close(fd);
  1187. // Clear service listeners
  1188. for (auto p : service_key_map) {
  1189. p.first->remove_listener(this);
  1190. }
  1191. active_control_conns--;
  1192. }