usr/replay/master.c

76     struct slave *sl;
260     struct slave *sl;
265     sl = pe->sl;
267     bulk_free(&sl->bt, bulk_id);
268     sl->pe = NULL;
288     struct slave *sl = &SlState.slaves[slavenum];
295     sl->b = b;
299     /* printf("assigned binding to %p\n", sl); */
688             struct slave *sl = SlState.slaves + sid;
690             if (sl->pe == NULL) {
691                 sl->pe = tg_pop(&TG);
692                 if (sl->pe == NULL) {
695                 dmsg("[MASTER] assigned pid:%d to sl:%d (stack_nr:%d completed:%d total:%d bytes:%zd)\n",
696                       sl->pe->pid, sid, TG.stack_nr, TG.pes_completed, TG.pes_nr, sl->pe->tes_size);
697                 sl->pe->sl = sl;
698                 slave_push_work(sl);
740         struct slave *sl = SlState.slaves + sid;
742             err = sl->b->tx_vtbl.slave_finish(sl->b, NOP_CONT);
761         struct slave *sl = SlState.slaves + sid;
763         err = sl->b->tx_vtbl.slave_print_stats(sl->b, NOP_CONT);
773 slave_push_work(struct slave *sl)
779     //dmsg("pushing work for slave: %ld (%p) pid:%d (completed:%d)\n", sl-slaves, sl, sl->pe->pid, sl->pe->completed);
780     bb = bulk_alloc(&sl->bt);
784     bulk_buf_copy(bb, sl->pe->tes, sl->pe->tes_size);
786     err = sl->b->tx_vtbl.new_task(sl->b, NOP_CONT, bulk_id, sl->pe->tes_size);
788         bulk_free(&sl->bt, bulk_id);
803         struct slave *sl = SlState.slaves + sid;
828         err = bulk_create(BULK_TOTAL_SIZE, BULK_BLOCK_SIZE, &sl->frame, &sl->bt);
830         err = sl->b->tx_vtbl.slave_init(sl->b, NOP_CONT, sl->frame, BULK_TOTAL_SIZE);
844 slave_push_work(struct slave *sl)
848     assert(sl->pe != NULL);
849     assert(sl->pe->tes_size > sl->sentbytes);
851     dmsg("TRYING TO SEND: %zd bytes for pid=%u\n", sl->pe->tes_size - sl->sentbytes, sl->pe->pid);
852     ret = send(sl->socket,
853                (char *)sl->pe->tes + sl->sentbytes,
854                sl->pe->tes_size - sl->sentbytes,
860     dmsg("SENT: %zd bytes for pid=%u\n", ret, sl->pe->pid);
861     sl->sentbytes += ret;
868         struct slave *sl = SlState.slaves + i;
869         close(sl->socket);
879         struct slave *sl = SlState.slaves + i;
881         if (sl->pe == NULL) {
887          if (sl->pe->tes_size > sl->sentbytes) {
888             slave_push_work(sl);
891         int r = recv(sl->socket, &rpid, sizeof(rpid), MSG_DONTWAIT);
898         assert(rpid == sl->pe->pid);
899         tg_complete(&TG, sl->pe);
900         sl->pe = NULL;
901         sl->sentbytes = 0;
917         struct slave *sl = &SlState.slaves[i];
919         sl->socket = socket(AF_INET, SOCK_STREAM, 0);
920         if (sl->socket == -1) {
933         ret = bind(sl->socket, (struct sockaddr *)&a, sizeof(a));
958         ret = connect(sl->socket, (struct sockaddr *)&sa, sizeof(sa));
964         sl->sentbytes = 0;
967         int sock_fl = fcntl(sl->socket, F_GETFD);
969         sock_fl = fcntl(sl->socket, F_SETFD, sock_fl);