controller.cc 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790
  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. unsigned int problems_remaining;
  90. struct evconnlistener *listener;
  91. ControllerState() : working(false), num_unsolved_subproblems(0),
  92. problems_remaining(0), 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. ZZ base_exp = PowerMod(ctrlstate.base, expon, ctrlstate.rho);
  336. if (base_exp == ctrlstate.target) {
  337. cout << "CORRECT in " << length_buf << "\n";
  338. } else {
  339. cout << "INCORRECT in " << length_buf << ":\n";
  340. cout << "base^exp = " << base_exp << "\n";
  341. cout << "target = " << ctrlstate.target << "\n";
  342. }
  343. ctrlstate.reset();
  344. }
  345. static unsigned short curproblemid = 0;
  346. // Take base and target mod f.factor, then decompose that into small
  347. // subproblems given our knowledge of the factors of phi(f.factor)
  348. static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
  349. const FactorDecomp &f)
  350. {
  351. vector<SubproblemProgress> ret;
  352. // Compute phi(factor)
  353. const int fveclen = f.fvec.length();
  354. ZZ phi = to_ZZ(2);
  355. for (int i = 0; i < fveclen; ++i) {
  356. phi *= f.fvec[i];
  357. }
  358. ZZ_p::init(f.factor);
  359. for (int i = 0; i < fveclen; ++i) {
  360. const ZZ& order = f.fvec[i];
  361. ZZ quotient = phi / order;
  362. ZZ_p subgroup_base = to_ZZ_p(rep(base));
  363. subgroup_base = power(subgroup_base, quotient);
  364. ZZ_p subgroup_target = to_ZZ_p(rep(target));
  365. subgroup_target = power(subgroup_target, quotient);
  366. if (subgroup_base == 1) {
  367. // The original base wasn't a generator of the whole group
  368. if (subgroup_target == 1) {
  369. // But the target is in the subgroup. Lucky us.
  370. continue;
  371. } else {
  372. ret.clear();
  373. return ret;
  374. }
  375. }
  376. ++curproblemid;
  377. ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
  378. rep(subgroup_target), f.factor, order));
  379. }
  380. return ret;
  381. }
  382. static int generate_problem(struct event_base *evbase)
  383. {
  384. // If there are no more problems to generate, close the listener
  385. if (ctrlstate.problems_remaining == 0) {
  386. evconnlistener_free(ctrlstate.listener);
  387. ctrlstate.dpnodes.free();
  388. ctrlstate.workers.free();
  389. return -1;
  390. }
  391. // If there's already a problem on the go, don't generate another one
  392. if (ctrlstate.working == true) {
  393. return -1;
  394. }
  395. ctrlstate.working = true;
  396. int num_subproblems_p = 0;
  397. int num_subproblems_q = 0;
  398. // Generate a DLP mod rho (in the large odd-order subgroup)
  399. ZZ_p::init(ctrlstate.rho);
  400. do {
  401. ZZ_p base = power(random_ZZ_p(), 2);
  402. ZZ_p target = power(random_ZZ_p(), 2);
  403. gettimeofday(&ctrlstate.started_working, NULL);
  404. ctrlstate.base = rep(base);
  405. ctrlstate.target = rep(target);
  406. // Decompose it mod p and mod q
  407. ctrlstate.subproblems_p = decomp(base, target, ctrlstate.p);
  408. ctrlstate.subproblems_q = decomp(base, target, ctrlstate.q);
  409. num_subproblems_p = ctrlstate.subproblems_p.size();
  410. num_subproblems_q = ctrlstate.subproblems_q.size();
  411. } while (num_subproblems_p == 0 || num_subproblems_q == 0);
  412. ctrlstate.num_unsolved_subproblems =
  413. num_subproblems_p + num_subproblems_q;
  414. schedule();
  415. --ctrlstate.problems_remaining;
  416. return 0;
  417. }
  418. typedef enum {
  419. CCSTATE_START,
  420. CCSTATE_DPWAITRESP,
  421. CCSTATE_DPLISTENING,
  422. CCSTATE_DPEXPON,
  423. CCSTATE_END
  424. } CCState;
  425. struct ControllerConnInfo {
  426. CCState state;
  427. ControllerConnInfo() : state(CCSTATE_DPWAITRESP) {}
  428. };
  429. static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
  430. {
  431. struct evbuffer *input = bufferevent_get_input(bev);
  432. ControllerConnInfo *info = (ControllerConnInfo *)ctx;
  433. unsigned char cmd[1];
  434. ZZ expon;
  435. while(1) {
  436. size_t len = evbuffer_get_length(input);
  437. switch (info->state) {
  438. case CCSTATE_START:
  439. case CCSTATE_DPWAITRESP:
  440. if (len < 1) return;
  441. bufferevent_read(bev, cmd, 1);
  442. switch (cmd[0]) {
  443. case 'L':
  444. info->state = CCSTATE_DPLISTENING;
  445. break;
  446. case 'E':
  447. info->state = CCSTATE_DPEXPON;
  448. break;
  449. default:
  450. /* Unknown DPnode command received */
  451. fprintf(stderr, "Unknown command in "
  452. "controller_dpnode_reader: "
  453. "%c\n", cmd[0]);
  454. info->state = CCSTATE_END;
  455. break;
  456. }
  457. break;
  458. case CCSTATE_DPLISTENING:
  459. // Read 6 bytes
  460. if (len < 6) return;
  461. unsigned char ipport[6];
  462. unsigned int DPip;
  463. unsigned short DPport;
  464. bufferevent_read(bev, ipport, 6);
  465. memmove(&DPip, ipport, 4);
  466. memmove(&DPport, ipport+4, 2);
  467. {
  468. #ifdef VERBOSE
  469. struct in_addr DPaddr = { DPip };
  470. fprintf(stderr, "DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
  471. #endif
  472. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  473. ctrlstate.dpnodes.working[bev]->ipports.push_back(
  474. IPPort(ipport));
  475. schedule();
  476. }
  477. }
  478. info->state = CCSTATE_DPWAITRESP;
  479. break;
  480. case CCSTATE_DPEXPON:
  481. // Read the subproblemid and the answer to the subproblem
  482. if (len < 2 + 3*sizeof(unsigned int)) return;
  483. unsigned char exponbytes[2 + 3*sizeof(unsigned int)];
  484. unsigned short problemid;
  485. bufferevent_read(bev, exponbytes, 2 + 3*sizeof(unsigned int));
  486. memmove(&problemid, exponbytes, 2);
  487. ZZFromBytes(expon, exponbytes+2, 3*sizeof(unsigned int));
  488. // Find the subproblem and check the answer
  489. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  490. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  491. if (spp->problemid == problemid &&
  492. spp->solved == false &&
  493. spp->target ==
  494. PowerMod(spp->base, expon, spp->modulus)) {
  495. // Subproblem solved!
  496. spp->solution = expon;
  497. spp->solved = true;
  498. spp->stop();
  499. ctrlstate.num_unsolved_subproblems--;
  500. if (ctrlstate.num_unsolved_subproblems == 0) {
  501. computation_complete();
  502. generate_problem(bufferevent_get_base(bev));
  503. }
  504. schedule();
  505. }
  506. }
  507. info->state = CCSTATE_DPWAITRESP;
  508. break;
  509. case CCSTATE_END:
  510. // Shut down the connection
  511. delete info;
  512. bufferevent_free(bev);
  513. return;
  514. }
  515. }
  516. }
  517. static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
  518. void *ctx)
  519. {
  520. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  521. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  522. fprintf(stderr, "Closing dpnode connection\n");
  523. if (ctrlstate.dpnodes.working.count(bev)) {
  524. // If we lose a dpnode from an active computation, the
  525. // computation is useless.
  526. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  527. ctrlstate.dpnodes.working.erase(bev);
  528. spp->dpnodes.erase(bev);
  529. spp->stop();
  530. } else {
  531. ctrlstate.dpnodes.idle.erase(bev);
  532. }
  533. delete info;
  534. bufferevent_free(bev);
  535. schedule();
  536. }
  537. }
  538. static void controller_worker_event_cb(struct bufferevent *bev, short events,
  539. void *ctx)
  540. {
  541. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  542. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  543. fprintf(stderr, "Closing worker connection\n");
  544. if (ctrlstate.workers.working.count(bev)) {
  545. SubproblemProgress *spp = ctrlstate.workers.working[bev];
  546. ctrlstate.workers.working.erase(bev);
  547. spp->workers.erase(bev);
  548. } else {
  549. ctrlstate.workers.idle.erase(bev);
  550. }
  551. delete info;
  552. bufferevent_free(bev);
  553. schedule();
  554. }
  555. }
  556. static void controller_event_cb(struct bufferevent *bev, short events,
  557. void *ctx)
  558. {
  559. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  560. fprintf(stderr, "Closing connection\n");
  561. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  562. delete info;
  563. bufferevent_free(bev);
  564. }
  565. }
  566. // We're just going to read a single byte that will tell us whether the
  567. // peer is a DPnode or a Worker
  568. static void controller_master_reader(struct bufferevent *bev, void *ctx)
  569. {
  570. struct evbuffer *input = bufferevent_get_input(bev);
  571. size_t len = evbuffer_get_length(input);
  572. if (len < 1) return;
  573. char indata[1];
  574. bufferevent_read(bev, indata, 1);
  575. switch(indata[0]) {
  576. case 'D':
  577. /* Add this DPnode to the list of available ones */
  578. ctrlstate.dpnodes.idle.insert(bev);
  579. bufferevent_setcb(bev, controller_dpnode_reader, NULL,
  580. controller_dpnode_event_cb, ctx);
  581. controller_dpnode_reader(bev, ctx);
  582. schedule();
  583. return;
  584. case 'W':
  585. ctrlstate.workers.idle.insert(bev);
  586. // We don't actually read anything from workers
  587. bufferevent_enable(bev, EV_WRITE);
  588. bufferevent_setcb(bev, NULL, NULL,
  589. controller_worker_event_cb, ctx);
  590. schedule();
  591. return;
  592. default:
  593. fprintf(stderr, "Unknown command in controller_master_reader: "
  594. "%c\n", indata[0]);
  595. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  596. delete info;
  597. bufferevent_free(bev);
  598. return;
  599. }
  600. }
  601. static void controller_accept_cb(struct evconnlistener *listener,
  602. evutil_socket_t fd, struct sockaddr *address, int socklen,
  603. void *ctx)
  604. {
  605. // Create the state of the new connection
  606. ControllerConnInfo *info = new ControllerConnInfo();
  607. // Create a bufferevent for the new connection
  608. struct event_base *base = evconnlistener_get_base(listener);
  609. struct bufferevent *bev = bufferevent_socket_new(
  610. base, fd, BEV_OPT_CLOSE_ON_FREE);
  611. bufferevent_setcb(bev, controller_master_reader, NULL,
  612. controller_event_cb, info);
  613. bufferevent_enable(bev, EV_READ|EV_WRITE);
  614. }
  615. // Create a new controller socket. bindport is the port to bind to (in
  616. // host byte order), or 0 if any port will do. ip and boundport are set
  617. // to the IP and port of the socket, in network byte order.
  618. static struct evconnlistener *controller_create(struct event_base *evbase,
  619. unsigned short bindport, unsigned int *ip, unsigned short *boundport)
  620. {
  621. return listener_create(evbase, bindport, controller_accept_cb, NULL,
  622. ip, boundport, false);
  623. }
  624. // Read the modulus (and the factorization of the modulus and its
  625. // totient) from the given file. "-" means cin. Returns true if
  626. // successful.
  627. static bool read_modulus(const char *filename)
  628. {
  629. if (strcmp(filename, "-")) {
  630. ifstream ins(filename);
  631. if (!ins.good()) return false;
  632. ins >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  633. ctrlstate.q.factor >> ctrlstate.q.fvec;
  634. ins.close();
  635. } else {
  636. cin >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  637. ctrlstate.q.factor >> ctrlstate.q.fvec;
  638. }
  639. return true;
  640. }
  641. int controller_main(const Worklist &worklist, unsigned short bindport)
  642. {
  643. // Initialize the prng with some randomness from the kernel
  644. unsigned char randbuf[1024];
  645. ifstream urand("/dev/urandom");
  646. urand.read((char *)randbuf, sizeof(randbuf));
  647. urand.close();
  648. ZZ randzz = ZZFromBytes(randbuf, sizeof(randbuf));
  649. SetSeed(randzz);
  650. if (worklist.size() == 0) {
  651. return 0;
  652. }
  653. // Read the modulus and the factorization of its totient from the
  654. // specified file
  655. if (!read_modulus(worklist[0].first)) {
  656. cerr << "Unable to read file " << worklist[0].first << "\n";
  657. return 1;
  658. }
  659. struct event_base *evbase = event_base_new();
  660. unsigned int myip;
  661. unsigned short myport;
  662. ctrlstate.listener = controller_create(evbase, bindport, &myip, &myport);
  663. #ifdef VERBOSE
  664. struct in_addr myaddr = { myip };
  665. printf("Bound to %s:%d\n", inet_ntoa(myaddr), ntohs(myport));
  666. #endif
  667. ctrlstate.problems_remaining = worklist[0].second;
  668. // Kick off the first problem to solve
  669. generate_problem(evbase);
  670. event_base_dispatch(evbase);
  671. return 0;
  672. }