control.cc 48 KB

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