#include #include #include #include #include #include #include #include #include "controller.h" #include "worker.h" #include "dpnode.h" NTL_CLIENT static int mpi_size; static int controllerfds[2]; // Close fds 3 and up, except for the one given (pass -1 to close them // all) static void close_highfds_except(int exceptfd) { // Find the max fd number struct rlimit limit; getrlimit(RLIMIT_NOFILE, &limit); for (int fd = 3; fd < limit.rlim_cur; ++fd) { if (fd != exceptfd) { // There's no ill effect from closing a non-open fd, so just // do it close(fd); } } } static void boundcb(const char *boundaddr, unsigned short boundport) { // Write the port and addr to the pipe write(controllerfds[1], &boundport, 2); write(controllerfds[1], boundaddr, strlen(boundaddr)); close(controllerfds[1]); } void desired_resources(const ZZ &order, unsigned short &desired_dpnodes, unsigned int &max_workers, unsigned int &dpfreq) { // How many DPnodes should we use for a problem of this size? desired_dpnodes = 1; // How many workers would we like to use? ZZ sorder = SqrRoot(order >> 46); if (NumBits(sorder) > 30) { // Just use all the workers we can find max_workers = 4294967295U; // 2^32 - 1 } else { max_workers = trunc_long(sorder,31) + 1; } // By default, 1 in 1000 points are distinguihed points. The // number in the next line is 2^32/1000 dpfreq = 4294967; if (order < 1000) { // Just make every point a DP dpfreq = 4294967295U; } else if (NumBits(order) < 27) { // The frequency of DPs should be 10/sqrt(order) to avoid // a DP-free cycle, so dpfreq = (10*2^32)/sqrt(order) ZZ f = (to_ZZ(10) << 32) / SqrRoot(order); dpfreq = trunc_long(f, 31); } } static pid_t fork_and_remember(vector &children) { pid_t pid = fork(); if (pid > 0) { children.push_back(pid); } return pid; } int main(int argc, char **argv) { // Init MPI MPI_Init(&argc, &argv); char hostname[257]; gethostname(hostname, 256); vector children; int rank = 0; int ret = 0; MPI_Comm_rank(MPI_COMM_WORLD, &rank); MPI_Comm_size(MPI_COMM_WORLD, &mpi_size); char boundportaddr[259]; if (rank == 0) { // Start the controller pipe(controllerfds); char fdenv[30]; sprintf(fdenv, "%d", controllerfds[1]); setenv("CONTROLLER_BOUNDCB_FD", fdenv, 1); if (fork_and_remember(children) == 0) { // Child; close the read half of the pipe and all other fds close_highfds_except(controllerfds[1]); execv("./controller", argv); return 1; } else { // Parent; close the write half of the pipe close(controllerfds[1]); int res; memset(boundportaddr, '\0', 259); res = read(controllerfds[0], boundportaddr, 2); if (res < 2) return 1; res = read(controllerfds[0], boundportaddr+2, 256); if (res < 1) return 1; close(controllerfds[0]); // std::cerr << "Parent reports controller bound to " << boundaddr << ":" << boundport << "\n"; } } MPI_Bcast(boundportaddr, 259, MPI_CHAR, 0, MPI_COMM_WORLD); unsigned short boundport; memmove(&boundport, boundportaddr, 2); char portstr[10]; sprintf(portstr, "%hu", boundport); const char *boundaddr = boundportaddr + 2; // The child will spawn two of its own children to be the // workers if (fork_and_remember(children) == 0) { close_highfds_except(-1); execl("./worker", "./worker", boundaddr, portstr, "0", NULL); return 1; } else if (fork_and_remember(children) == 0) { close_highfds_except(-1); execl("./worker", "./worker", boundaddr, portstr, "1", NULL); return 1; } else if (fork_and_remember(children) == 0) { // And a dpnode close_highfds_except(-1); execl("./dpnode", "./dpnode", boundaddr, portstr, NULL); return 1; } // Now wait for all the children vector::iterator pidit; for (pidit = children.begin(); pidit != children.end(); ++pidit) { int status; waitpid(*pidit, &status, 0); } MPI_Finalize(); return ret; }