controller.cc 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824
  1. extern "C" {
  2. #include <event2/listener.h>
  3. #include <event2/bufferevent.h>
  4. #include <event2/buffer.h>
  5. }
  6. #include <NTL/vec_ZZ.h>
  7. #include <NTL/ZZ.h>
  8. #include <NTL/ZZ_p.h>
  9. #include <sys/socket.h>
  10. #include <netinet/in.h>
  11. #include <arpa/inet.h>
  12. #include <sys/time.h>
  13. #include <fstream>
  14. #include <vector>
  15. #include <set>
  16. #include <map>
  17. #include <string.h>
  18. #include "evutils.h"
  19. #include "subproblem.h"
  20. #include "controller.h"
  21. NTL_CLIENT
  22. // #undef VERBOSE
  23. struct SubproblemProgress;
  24. typedef std::set<struct bufferevent *> BESet;
  25. typedef std::map<struct bufferevent *, SubproblemProgress *> BEMap;
  26. static void besetdump(const BESet &bes, ostream &os)
  27. {
  28. BESet::const_iterator besit;
  29. os << hex << " ";
  30. for (besit = bes.begin(); besit != bes.end(); ++besit) {
  31. os << *besit << " ";
  32. }
  33. os << dec << "\n";
  34. }
  35. static void besetfree(BESet &bes)
  36. {
  37. BESet::iterator besit;
  38. for (besit = bes.begin(); besit != bes.end(); ++besit) {
  39. bufferevent_free(*besit);
  40. }
  41. bes.clear();
  42. }
  43. static void bemapdump(const BEMap &bem, ostream &os)
  44. {
  45. BEMap::const_iterator bemit;
  46. os << hex << " ";
  47. for (bemit = bem.begin(); bemit != bem.end(); ++bemit) {
  48. os << bemit->first << "->" << bemit->second << " ";
  49. }
  50. os << dec << "\n";
  51. }
  52. static void bemapfree(BEMap &bem)
  53. {
  54. BEMap::iterator bemit;
  55. for (bemit = bem.begin(); bemit != bem.end(); ++bemit) {
  56. bufferevent_free(bemit->first);
  57. }
  58. bem.clear();
  59. }
  60. struct Statuses {
  61. BESet idle;
  62. BEMap working;
  63. // Dump the state for debug purposes
  64. void dump(ostream &os) const {
  65. os << " idle (" << idle.size() << "):\n";
  66. besetdump(idle, os);
  67. os << " working (" << working.size() << "):\n";
  68. bemapdump(working, os);
  69. }
  70. void free(void) {
  71. besetfree(idle);
  72. bemapfree(working);
  73. }
  74. };
  75. struct FactorDecomp {
  76. ZZ factor;
  77. vec_ZZ fvec;
  78. };
  79. static void vsppdump(const vector<SubproblemProgress> &spv, ostream &os);
  80. static struct ControllerState {
  81. ZZ rho;
  82. FactorDecomp p, q;
  83. ZZ base, target;
  84. bool working;
  85. struct timeval started_working;
  86. vector<SubproblemProgress> subproblems_p, subproblems_q;
  87. Statuses dpnodes, workers;
  88. unsigned int num_unsolved_subproblems;
  89. Worklist worklist;
  90. struct evconnlistener *listener;
  91. ControllerState() : working(false), num_unsolved_subproblems(0),
  92. listener(NULL) {}
  93. // Reset the state for a new problem with the same modulus
  94. void reset(void) {
  95. base = 0;
  96. target = 0;
  97. working = false;
  98. subproblems_p.clear();
  99. subproblems_q.clear();
  100. num_unsolved_subproblems = 0;
  101. started_working.tv_sec = 0;
  102. started_working.tv_usec = 0;
  103. }
  104. // Dump the state for debug purposes
  105. void dump(ostream &os) const {
  106. if (!working) {
  107. os << "Not working\n";
  108. return;
  109. }
  110. os << "P:\n";
  111. vsppdump(subproblems_p, os);
  112. os << "Q:\n";
  113. vsppdump(subproblems_q, os);
  114. os << "dpnodes:\n";
  115. dpnodes.dump(os);
  116. os << "workers:\n";
  117. workers.dump(os);
  118. }
  119. } ctrlstate;
  120. struct IPPort {
  121. unsigned char ipport[6];
  122. IPPort(unsigned char *ipp) {
  123. memmove(ipport, ipp, 6);
  124. }
  125. void dump(ostream &os) const {
  126. os << int(ipport[0]) << "." << int(ipport[1]) << "." <<
  127. int(ipport[2]) << "." << int(ipport[3]) << ":" <<
  128. ((ipport[4] << 8) + ipport[5]) << " ";
  129. }
  130. };
  131. typedef vector<IPPort> IPPortSet;
  132. static void ipportsetdump(const IPPortSet &ipps, ostream &os)
  133. {
  134. IPPortSet::const_iterator ippsit;
  135. os << " ";
  136. for (ippsit = ipps.begin(); ippsit != ipps.end(); ++ippsit) {
  137. ippsit->dump(os);
  138. }
  139. os << "\n";
  140. }
  141. void desired_resources(const ZZ &order, unsigned short &desired_dpnodes,
  142. unsigned int &max_workers, unsigned int &dpfreq);
  143. struct SubproblemProgress : Subproblem {
  144. // The sets of dpnodes and workers currently working on this subproblem
  145. BESet dpnodes, workers;
  146. // The dpnode IPPorts registered for this subproblem
  147. IPPortSet ipports;
  148. // The desired number of DPnodes for this subproblem
  149. unsigned short desired_dpnodes;
  150. // The maximum number of workers useful for this subproblem
  151. unsigned int max_workers;
  152. // Have we found a solution?
  153. bool solved;
  154. // The solution, if found.
  155. ZZ solution;
  156. SubproblemProgress(unsigned short id, const ZZ &b, const ZZ &t,
  157. const ZZ &m, const ZZ &o) :
  158. // By default, 1 in 1000 points are distinguihed points. The
  159. // number in the next line is 2^32/1000
  160. Subproblem(id, b, t, m, o, 4294967), solved(false) {
  161. desired_resources(order, desired_dpnodes, max_workers, dpfreq);
  162. }
  163. // Stop all dpnodes and workers
  164. void stop(void) {
  165. BESet::iterator iter;
  166. unsigned char stopcmd[1] = { 'S' };
  167. for (BESet::iterator iter = dpnodes.begin(); iter != dpnodes.end();
  168. ++iter) {
  169. bufferevent_write(*iter, stopcmd, 1);
  170. ctrlstate.dpnodes.working.erase(*iter);
  171. ctrlstate.dpnodes.idle.insert(*iter);
  172. }
  173. for (BESet::iterator iter = workers.begin(); iter != workers.end();
  174. ++iter) {
  175. bufferevent_write(*iter, stopcmd, 1);
  176. ctrlstate.workers.working.erase(*iter);
  177. ctrlstate.workers.idle.insert(*iter);
  178. }
  179. dpnodes.clear();
  180. workers.clear();
  181. ipports.clear();
  182. }
  183. // Dump for debugging purposes
  184. void dump(ostream &os) const {
  185. os << " Subproblem " << problemid << "\n";
  186. os << " dpnodes (" << dpnodes.size() << "):\n";
  187. besetdump(dpnodes, os);
  188. os << " workers (" << workers.size() << "):\n";
  189. besetdump(workers, os);
  190. os << " ipports (" << ipports.size() << "):\n";
  191. ipportsetdump(ipports, os);
  192. if (solved) {
  193. os << " solution: " << solution << "\n\n";
  194. }
  195. }
  196. void worker_write(struct bufferevent *bev) {
  197. bev_write(bev);
  198. unsigned short num_ipports = ipports.size();
  199. bufferevent_write(bev, &num_ipports, 2);
  200. for (unsigned short i = 0; i < num_ipports; ++i) {
  201. bufferevent_write(bev, ipports[i].ipport, 6);
  202. }
  203. #ifdef VERBOSE
  204. cerr << "Added worker " << bev << " to subproblem "
  205. << problemid << "\n";
  206. #endif
  207. }
  208. };
  209. // Dump the state for debug purposes
  210. static void vsppdump(const vector<SubproblemProgress> &spv, ostream &os)
  211. {
  212. vector<SubproblemProgress>::const_iterator spiter;
  213. int count = 0;
  214. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  215. ++count;
  216. os << " " << count << ":\n";
  217. spiter->dump(os);
  218. }
  219. os << "\n";
  220. }
  221. // Find a subproblem in the given vector that could use another DPnode,
  222. // and give it one of the idle ones. Only allocate it to a subproblem
  223. // with no current DPnodes if consider_empty is true.
  224. static void find_subproblem_for_dpnode(vector<SubproblemProgress> &spv,
  225. bool consider_empty)
  226. {
  227. vector<SubproblemProgress>::iterator spiter;
  228. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  229. if (spiter->solved) continue;
  230. if (spiter->dpnodes.size() == 0 && consider_empty == false) continue;
  231. // How many DPnodes would we like to have for this subproblem?
  232. while (spiter->dpnodes.size() < spiter->desired_dpnodes &&
  233. ctrlstate.dpnodes.idle.size() > 0) {
  234. // Get the first idle DPnode
  235. BESet::iterator beviter = ctrlstate.dpnodes.idle.begin();
  236. struct bufferevent *firstbev = *beviter;
  237. // Allocate it to the subproblem
  238. spiter->dpnodes.insert(firstbev);
  239. ctrlstate.dpnodes.working[firstbev] = &(*spiter);
  240. ctrlstate.dpnodes.idle.erase(firstbev);
  241. // Tell it to start listening for DPs
  242. spiter->bev_write(firstbev);
  243. }
  244. }
  245. }
  246. // Find a subproblem in the given vector that has all of its DPnodes and
  247. // could use another worker, and give it one of the idle ones.
  248. static void find_subproblem_for_worker(vector<SubproblemProgress> &spv)
  249. {
  250. vector<SubproblemProgress>::iterator spiter;
  251. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  252. if (spiter->solved) continue;
  253. while (spiter->ipports.size() == spiter->desired_dpnodes &&
  254. spiter->workers.size() < spiter->max_workers &&
  255. ctrlstate.workers.idle.size() > 0) {
  256. // Get the first idle worker
  257. BESet::iterator beviter = ctrlstate.workers.idle.begin();
  258. struct bufferevent *firstbev = *beviter;
  259. // Allocate it to the subproblem
  260. spiter->workers.insert(firstbev);
  261. ctrlstate.workers.working[firstbev] = &(*spiter);
  262. ctrlstate.workers.idle.erase(firstbev);
  263. // Tell it to start working on the subproblem
  264. spiter->worker_write(firstbev);
  265. }
  266. }
  267. }
  268. // See if there are any idle DPnodes or workers we can put to use
  269. static void schedule(void)
  270. {
  271. // Check the DPnodes
  272. // Iterate through the subproblems, looking for one that can use
  273. // another DPnode. First look for subproblems that already have
  274. // some, but not all, of their DPnodes
  275. if (ctrlstate.dpnodes.idle.size() > 0) {
  276. find_subproblem_for_dpnode(ctrlstate.subproblems_p, false);
  277. }
  278. if (ctrlstate.dpnodes.idle.size() > 0) {
  279. find_subproblem_for_dpnode(ctrlstate.subproblems_q, false);
  280. }
  281. // If there are still more dpnodes to place, start assigning them to
  282. // subproblems with no current dpnodes
  283. if (ctrlstate.dpnodes.idle.size() > 0) {
  284. find_subproblem_for_dpnode(ctrlstate.subproblems_p, true);
  285. }
  286. if (ctrlstate.dpnodes.idle.size() > 0) {
  287. find_subproblem_for_dpnode(ctrlstate.subproblems_q, true);
  288. }
  289. // Check the workers
  290. // Iterate through the subproblems, looking for one that can use
  291. // another worker.
  292. if (ctrlstate.workers.idle.size() > 0) {
  293. find_subproblem_for_worker(ctrlstate.subproblems_p);
  294. }
  295. if (ctrlstate.workers.idle.size() > 0) {
  296. find_subproblem_for_worker(ctrlstate.subproblems_q);
  297. }
  298. // cerr << "After schedule:\n"; ctrlstate.dump(cerr);
  299. }
  300. static ZZ computation_complete_p(const vector<SubproblemProgress> &v)
  301. {
  302. ZZ curmodulus, curexp;
  303. curmodulus = 2;
  304. curexp = 0;
  305. vector<SubproblemProgress>::const_iterator vit;
  306. for(vit = v.begin(); vit != v.end(); ++vit) {
  307. CRT(curexp, curmodulus, vit->solution, vit->order);
  308. }
  309. if (curexp < 0) {
  310. curexp += curmodulus;
  311. }
  312. return curexp;
  313. }
  314. // All subproblems are solved. Combine the results.
  315. static void computation_complete(void)
  316. {
  317. ZZ exp_p = computation_complete_p(ctrlstate.subproblems_p);
  318. ZZ exp_q = computation_complete_p(ctrlstate.subproblems_q);
  319. ZZ pm1 = (ctrlstate.p.factor - 1)/2;
  320. ZZ qm1 = (ctrlstate.q.factor - 1)/2;
  321. CRT(exp_p, pm1, exp_q, qm1);
  322. if (exp_p < 0) {
  323. exp_p += pm1;
  324. }
  325. ZZ& expon = exp_p;
  326. struct timeval ended_working;
  327. gettimeofday(&ended_working, NULL);
  328. unsigned long long computation_length_ms =
  329. (ended_working.tv_sec - ctrlstate.started_working.tv_sec) * 1000 +
  330. (ended_working.tv_usec - ctrlstate.started_working.tv_usec) / 1000;
  331. char length_buf[50];
  332. sprintf(length_buf, "%lld.%03lld s", computation_length_ms / 1000,
  333. computation_length_ms % 1000);
  334. cout << "expon = " << expon << "\n";
  335. cout << ctrlstate.worklist[0].first << " ";
  336. ZZ base_exp = PowerMod(ctrlstate.base, expon, ctrlstate.rho);
  337. if (base_exp == ctrlstate.target) {
  338. cout << "CORRECT in " << length_buf << "\n";
  339. } else {
  340. cout << "INCORRECT in " << length_buf << ":\n";
  341. cout << "base^exp = " << base_exp << "\n";
  342. cout << "target = " << ctrlstate.target << "\n";
  343. }
  344. ctrlstate.reset();
  345. }
  346. static unsigned short curproblemid = 0;
  347. // Take base and target mod f.factor, then decompose that into small
  348. // subproblems given our knowledge of the factors of phi(f.factor)
  349. static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
  350. const FactorDecomp &f)
  351. {
  352. vector<SubproblemProgress> ret;
  353. // Compute phi(factor)
  354. const int fveclen = f.fvec.length();
  355. ZZ phi = to_ZZ(2);
  356. for (int i = 0; i < fveclen; ++i) {
  357. phi *= f.fvec[i];
  358. }
  359. ZZ_p::init(f.factor);
  360. for (int i = 0; i < fveclen; ++i) {
  361. const ZZ& order = f.fvec[i];
  362. ZZ quotient = phi / order;
  363. ZZ_p subgroup_base = to_ZZ_p(rep(base));
  364. subgroup_base = power(subgroup_base, quotient);
  365. ZZ_p subgroup_target = to_ZZ_p(rep(target));
  366. subgroup_target = power(subgroup_target, quotient);
  367. if (subgroup_base == 1) {
  368. // The original base wasn't a generator of the whole group
  369. if (subgroup_target == 1) {
  370. // But the target is in the subgroup. Lucky us.
  371. continue;
  372. } else {
  373. ret.clear();
  374. return ret;
  375. }
  376. }
  377. ++curproblemid;
  378. ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
  379. rep(subgroup_target), f.factor, order));
  380. }
  381. return ret;
  382. }
  383. // Read the modulus (and the factorization of the modulus and its
  384. // totient) from the given file. "-" means cin. Returns true if
  385. // successful.
  386. static bool read_modulus(const char *filename)
  387. {
  388. if (strcmp(filename, "-")) {
  389. ifstream ins(filename);
  390. if (!ins.good()) return false;
  391. ins >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  392. ctrlstate.q.factor >> ctrlstate.q.fvec;
  393. ins.close();
  394. } else {
  395. cin >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  396. ctrlstate.q.factor >> ctrlstate.q.fvec;
  397. }
  398. return true;
  399. }
  400. static int generate_problem(struct event_base *evbase)
  401. {
  402. while (ctrlstate.worklist[0].second == 0) {
  403. ctrlstate.worklist.erase(ctrlstate.worklist.begin());
  404. if (ctrlstate.worklist.size() > 0) {
  405. read_modulus(ctrlstate.worklist[0].first);
  406. } else {
  407. break;
  408. }
  409. }
  410. // If there are no more problems to generate, close the listener
  411. if (ctrlstate.worklist.size() == 0) {
  412. evconnlistener_free(ctrlstate.listener);
  413. ctrlstate.dpnodes.free();
  414. ctrlstate.workers.free();
  415. return -1;
  416. }
  417. // If there's already a problem on the go, don't generate another one
  418. if (ctrlstate.working == true) {
  419. return -1;
  420. }
  421. ctrlstate.working = true;
  422. int num_subproblems_p = 0;
  423. int num_subproblems_q = 0;
  424. // Generate a DLP mod rho (in the large odd-order subgroup)
  425. ZZ_p::init(ctrlstate.rho);
  426. do {
  427. ZZ_p base = power(random_ZZ_p(), 2);
  428. ZZ_p target = power(random_ZZ_p(), 2);
  429. gettimeofday(&ctrlstate.started_working, NULL);
  430. ctrlstate.base = rep(base);
  431. ctrlstate.target = rep(target);
  432. // Decompose it mod p and mod q
  433. ctrlstate.subproblems_p = decomp(base, target, ctrlstate.p);
  434. ctrlstate.subproblems_q = decomp(base, target, ctrlstate.q);
  435. num_subproblems_p = ctrlstate.subproblems_p.size();
  436. num_subproblems_q = ctrlstate.subproblems_q.size();
  437. } while (num_subproblems_p == 0 || num_subproblems_q == 0);
  438. ctrlstate.num_unsolved_subproblems =
  439. num_subproblems_p + num_subproblems_q;
  440. schedule();
  441. --(ctrlstate.worklist[0].second);
  442. return 0;
  443. }
  444. typedef enum {
  445. CCSTATE_START,
  446. CCSTATE_DPWAITRESP,
  447. CCSTATE_DPLISTENING,
  448. CCSTATE_DPEXPON,
  449. CCSTATE_END
  450. } CCState;
  451. struct ControllerConnInfo {
  452. CCState state;
  453. ControllerConnInfo() : state(CCSTATE_DPWAITRESP) {}
  454. };
  455. static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
  456. {
  457. struct evbuffer *input = bufferevent_get_input(bev);
  458. ControllerConnInfo *info = (ControllerConnInfo *)ctx;
  459. unsigned char cmd[1];
  460. ZZ expon;
  461. while(1) {
  462. size_t len = evbuffer_get_length(input);
  463. switch (info->state) {
  464. case CCSTATE_START:
  465. case CCSTATE_DPWAITRESP:
  466. if (len < 1) return;
  467. bufferevent_read(bev, cmd, 1);
  468. switch (cmd[0]) {
  469. case 'L':
  470. info->state = CCSTATE_DPLISTENING;
  471. break;
  472. case 'E':
  473. info->state = CCSTATE_DPEXPON;
  474. break;
  475. default:
  476. /* Unknown DPnode command received */
  477. fprintf(stderr, "Unknown command in "
  478. "controller_dpnode_reader: "
  479. "%c\n", cmd[0]);
  480. info->state = CCSTATE_END;
  481. break;
  482. }
  483. break;
  484. case CCSTATE_DPLISTENING:
  485. // Read 6 bytes
  486. if (len < 6) return;
  487. unsigned char ipport[6];
  488. unsigned int DPip;
  489. unsigned short DPport;
  490. bufferevent_read(bev, ipport, 6);
  491. memmove(&DPip, ipport, 4);
  492. memmove(&DPport, ipport+4, 2);
  493. {
  494. #ifdef VERBOSE
  495. struct in_addr DPaddr = { DPip };
  496. fprintf(stderr, "DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
  497. #endif
  498. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  499. ctrlstate.dpnodes.working[bev]->ipports.push_back(
  500. IPPort(ipport));
  501. schedule();
  502. }
  503. }
  504. info->state = CCSTATE_DPWAITRESP;
  505. break;
  506. case CCSTATE_DPEXPON:
  507. // Read the subproblemid and the answer to the subproblem
  508. if (len < 2 + 3*sizeof(unsigned int)) return;
  509. unsigned char exponbytes[2 + 3*sizeof(unsigned int)];
  510. unsigned short problemid;
  511. bufferevent_read(bev, exponbytes, 2 + 3*sizeof(unsigned int));
  512. memmove(&problemid, exponbytes, 2);
  513. ZZFromBytes(expon, exponbytes+2, 3*sizeof(unsigned int));
  514. // Find the subproblem and check the answer
  515. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  516. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  517. if (spp->problemid == problemid &&
  518. spp->solved == false &&
  519. spp->target ==
  520. PowerMod(spp->base, expon, spp->modulus)) {
  521. // Subproblem solved!
  522. spp->solution = expon;
  523. spp->solved = true;
  524. spp->stop();
  525. ctrlstate.num_unsolved_subproblems--;
  526. if (ctrlstate.num_unsolved_subproblems == 0) {
  527. computation_complete();
  528. generate_problem(bufferevent_get_base(bev));
  529. }
  530. schedule();
  531. }
  532. }
  533. info->state = CCSTATE_DPWAITRESP;
  534. break;
  535. case CCSTATE_END:
  536. // Shut down the connection
  537. delete info;
  538. bufferevent_free(bev);
  539. return;
  540. }
  541. }
  542. }
  543. static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
  544. void *ctx)
  545. {
  546. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  547. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  548. fprintf(stderr, "Closing dpnode connection\n");
  549. if (ctrlstate.dpnodes.working.count(bev)) {
  550. // If we lose a dpnode from an active computation, the
  551. // computation is useless.
  552. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  553. ctrlstate.dpnodes.working.erase(bev);
  554. spp->dpnodes.erase(bev);
  555. spp->stop();
  556. } else {
  557. ctrlstate.dpnodes.idle.erase(bev);
  558. }
  559. delete info;
  560. bufferevent_free(bev);
  561. schedule();
  562. }
  563. }
  564. static void controller_worker_event_cb(struct bufferevent *bev, short events,
  565. void *ctx)
  566. {
  567. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  568. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  569. fprintf(stderr, "Closing worker connection\n");
  570. if (ctrlstate.workers.working.count(bev)) {
  571. SubproblemProgress *spp = ctrlstate.workers.working[bev];
  572. ctrlstate.workers.working.erase(bev);
  573. spp->workers.erase(bev);
  574. } else {
  575. ctrlstate.workers.idle.erase(bev);
  576. }
  577. delete info;
  578. bufferevent_free(bev);
  579. schedule();
  580. }
  581. }
  582. static void controller_event_cb(struct bufferevent *bev, short events,
  583. void *ctx)
  584. {
  585. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  586. fprintf(stderr, "Closing connection\n");
  587. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  588. delete info;
  589. bufferevent_free(bev);
  590. }
  591. }
  592. // We're just going to read a single byte that will tell us whether the
  593. // peer is a DPnode or a Worker
  594. static void controller_master_reader(struct bufferevent *bev, void *ctx)
  595. {
  596. struct evbuffer *input = bufferevent_get_input(bev);
  597. size_t len = evbuffer_get_length(input);
  598. if (len < 1) return;
  599. char indata[1];
  600. bufferevent_read(bev, indata, 1);
  601. switch(indata[0]) {
  602. case 'D':
  603. /* Add this DPnode to the list of available ones */
  604. ctrlstate.dpnodes.idle.insert(bev);
  605. bufferevent_setcb(bev, controller_dpnode_reader, NULL,
  606. controller_dpnode_event_cb, ctx);
  607. controller_dpnode_reader(bev, ctx);
  608. schedule();
  609. return;
  610. case 'W':
  611. ctrlstate.workers.idle.insert(bev);
  612. // We don't actually read anything from workers
  613. bufferevent_enable(bev, EV_WRITE);
  614. bufferevent_setcb(bev, NULL, NULL,
  615. controller_worker_event_cb, ctx);
  616. schedule();
  617. return;
  618. default:
  619. fprintf(stderr, "Unknown command in controller_master_reader: "
  620. "%c\n", indata[0]);
  621. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  622. delete info;
  623. bufferevent_free(bev);
  624. return;
  625. }
  626. }
  627. static void controller_accept_cb(struct evconnlistener *listener,
  628. evutil_socket_t fd, struct sockaddr *address, int socklen,
  629. void *ctx)
  630. {
  631. // Create the state of the new connection
  632. ControllerConnInfo *info = new ControllerConnInfo();
  633. // Create a bufferevent for the new connection
  634. struct event_base *base = evconnlistener_get_base(listener);
  635. struct bufferevent *bev = bufferevent_socket_new(
  636. base, fd, BEV_OPT_CLOSE_ON_FREE);
  637. bufferevent_setcb(bev, controller_master_reader, NULL,
  638. controller_event_cb, info);
  639. bufferevent_enable(bev, EV_READ|EV_WRITE);
  640. }
  641. // Create a new controller socket. bindport is the port to bind to (in
  642. // host byte order), or 0 if any port will do. ip and boundport are set
  643. // to the IP and port of the socket, in network byte order.
  644. static struct evconnlistener *controller_create(struct event_base *evbase,
  645. unsigned short bindport, unsigned int *ip, unsigned short *boundport)
  646. {
  647. return listener_create(evbase, bindport, controller_accept_cb, NULL,
  648. ip, boundport, false);
  649. }
  650. int controller_parse_args(int argc, char **argv, unsigned short &bindport,
  651. Worklist &worklist)
  652. {
  653. bindport = 0;
  654. if (argc > 2 && !strncmp(argv[1], "-p", 2)) {
  655. // A port number was specified
  656. bindport = strtoul(argv[2], NULL, 10);
  657. argc -= 2;
  658. argv += 2;
  659. }
  660. if (argc < 3 || (argc % 2 == 0)) {
  661. return 1;
  662. }
  663. for (int i=1;i<argc;i+=2) {
  664. worklist.push_back(Workentry(argv[i], strtoul(argv[i+1], NULL, 10)));
  665. }
  666. return 0;
  667. }
  668. int controller_main(const Worklist &worklist, unsigned short bindport,
  669. void (*boundcb)(const char *boundaddr, unsigned short boundport))
  670. {
  671. // Initialize the prng with some randomness from the kernel
  672. unsigned char randbuf[1024];
  673. ifstream urand("/dev/urandom");
  674. urand.read((char *)randbuf, sizeof(randbuf));
  675. urand.close();
  676. ZZ randzz = ZZFromBytes(randbuf, sizeof(randbuf));
  677. SetSeed(randzz);
  678. if (worklist.size() == 0) {
  679. return 0;
  680. }
  681. ctrlstate.worklist = worklist;
  682. // Read the modulus and the factorization of its totient from the
  683. // specified file
  684. if (!read_modulus(worklist[0].first)) {
  685. cerr << "Unable to read file " << worklist[0].first << "\n";
  686. return 1;
  687. }
  688. struct event_base *evbase = event_base_new();
  689. unsigned int myip;
  690. unsigned short myport;
  691. ctrlstate.listener = controller_create(evbase, bindport, &myip, &myport);
  692. if (ctrlstate.listener && boundcb) {
  693. struct in_addr myaddr = { myip };
  694. boundcb(inet_ntoa(myaddr), ntohs(myport));
  695. }
  696. // Kick off the first problem to solve
  697. generate_problem(evbase);
  698. event_base_dispatch(evbase);
  699. return 0;
  700. }