--- libaitrpc/src/srv.c 2015/06/30 14:11:55 1.26.2.10 +++ libaitrpc/src/srv.c 2015/07/22 15:07:10 1.27.2.6 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.26.2.10 2015/06/30 14:11:55 misho Exp $ +* $Id: srv.c,v 1.27.2.6 2015/07/22 15:07:10 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -82,7 +82,6 @@ static sched_task_func_t cbProto[SOCK_MAX_SUPPORT][4] static volatile uintptr_t _glSigArg = 0; - void rpc_freeCli(rpc_cli_t * __restrict c) { @@ -197,14 +196,14 @@ txPacket(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else if (rpc_pktFreeSpace(c) > s->srv_netbuf) { rpc_SetErr(EMSGSIZE, "Message too long"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); /* Go Encapsulate variables */ @@ -214,8 +213,8 @@ txPacket(sched_task_t *task) rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else wlen += ret; } @@ -263,8 +262,8 @@ execCall(sched_task_t *task) rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError()); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); taskExit(task, NULL); } } else @@ -274,15 +273,15 @@ execCall(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { /* if client doesn't want reply */ - rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr)); + RPC_SET_RETURN(rpc, rpc_srv_execCall(c, rpc, f->func_name, arr)); if (rpc->call_rep.ret == htonl(-1)) { if (!rpc->call_rep.eno) { LOGERR; - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_ERRNO(rpc, rpc_Errno); } rpc->call_argc ^= rpc->call_argc; ait_freeVars(&c->cli_vars); @@ -339,8 +338,8 @@ rxPacket(sched_task_t *task) rpc_SetErr(ERPCMISMATCH, "Short RPC packet"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); goto err; } else len = ntohl(rpc->call_len); @@ -348,8 +347,8 @@ rxPacket(sched_task_t *task) rpc_SetErr(ERPCMISMATCH, "Short RPC packet"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); goto err; } @@ -367,8 +366,8 @@ rxPacket(sched_task_t *task) rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); goto err; } #endif @@ -378,8 +377,8 @@ rxPacket(sched_task_t *task) rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); goto err; } @@ -464,14 +463,14 @@ txUDPPacket(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else if (rpc_pktFreeSpace(c) > s->srv_netbuf) { rpc_SetErr(EMSGSIZE, "Message too long"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); /* Go Encapsulate variables */ @@ -481,8 +480,8 @@ txUDPPacket(sched_task_t *task) rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else wlen += ret; } @@ -525,7 +524,7 @@ rxUDPPacket(sched_task_t *task) /* receive connect packet */ AIT_SET_BUF(&b, NULL, srv->srv_netbuf); - salen = sa.ss.ss_len = sizeof(sockaddr_t); + salen = sa.ss.ss_len = (u_char) MIN(sizeof(sockaddr_t), 0xff); rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen); rpc = (struct tagRPCCall*) AIT_GET_BUF(&b); if (rlen < sizeof(struct tagRPCCall)) @@ -604,14 +603,14 @@ txRAWPacket(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else if (rpc_pktFreeSpace(c) > s->srv_netbuf) { rpc_SetErr(EMSGSIZE, "Message too long"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); /* Go Encapsulate variables */ @@ -621,8 +620,8 @@ txRAWPacket(sched_task_t *task) rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else wlen += ret; } @@ -665,7 +664,7 @@ rxRAWPacket(sched_task_t *task) /* receive connect packet */ AIT_SET_BUF(&b, NULL, srv->srv_netbuf); - salen = sa.ss.ss_len = sizeof(sockaddr_t); + salen = sa.ss.ss_len = (u_char) MIN(sizeof(sockaddr_t), 0xff); rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen); if (sa.sa.sa_family == AF_INET) { struct ip *h; @@ -765,14 +764,14 @@ txBPFPacket(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else if (rpc_pktFreeSpace(c) > s->srv_netbuf) { rpc_SetErr(EMSGSIZE, "Message too long"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); /* Go Encapsulate variables */ @@ -782,8 +781,8 @@ txBPFPacket(sched_task_t *task) rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else wlen += ret; } @@ -928,14 +927,14 @@ txEXTPacket(sched_task_t *task) rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else if (rpc_pktFreeSpace(c) > s->srv_netbuf) { rpc_SetErr(EMSGSIZE, "Message too long"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else { rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); /* Go Encapsulate variables */ @@ -945,8 +944,8 @@ txEXTPacket(sched_task_t *task) rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(rpc_Errno); + RPC_SET_RETURN(rpc, -1); + RPC_SET_ERRNO(rpc, rpc_Errno); } else wlen += ret; } @@ -1118,7 +1117,7 @@ rxBLOB(sched_task_t *task) if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) { rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var)); blob.hdr_cmd = no; - blob.hdr_ret = RPC_ERROR(-1); + RPC_SET_BLOB_RET(&blob, -1); break; } else blob.hdr_len = htonl(b->blob_len); @@ -1129,7 +1128,7 @@ rxBLOB(sched_task_t *task) rpc_srv_blobUnmap(b); } else { blob.hdr_cmd = error; - blob.hdr_ret = RPC_ERROR(-1); + RPC_SET_BLOB_RET(&blob, -1); } break; case set: @@ -1143,19 +1142,19 @@ rxBLOB(sched_task_t *task) rpc_srv_blobUnmap(b); } else { blob.hdr_cmd = error; - blob.hdr_ret = RPC_ERROR(-1); + RPC_SET_BLOB_RET(&blob, -1); } break; case unset: if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) { blob.hdr_cmd = error; - blob.hdr_ret = RPC_ERROR(-1); + RPC_SET_BLOB_RET(&blob, -1); } break; default: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd); blob.hdr_cmd = error; - blob.hdr_ret = RPC_ERROR(-1); + RPC_SET_BLOB_RET(&blob, -1); } end: @@ -1168,7 +1167,11 @@ end: static void * flushBLOB(sched_task_t *task) { +#ifdef atomic_load_acq_ptr uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg); +#else + uintptr_t sigArg = *((volatile uintptr_t*) &_glSigArg); +#endif rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task); rpc_blob_t *b, *tmp; @@ -1376,10 +1379,10 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) srv->srv_blob.kill = 1; - schedEnd(&srv->srv_blob.root); - if (srv->srv_blob.server.cli_sa.sa.sa_family == AF_LOCAL) unlink(srv->srv_blob.server.cli_sa.sun.sun_path); + + schedEnd(&srv->srv_blob.root); } /* @@ -1410,7 +1413,11 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) /* disabled kqueue support in libaitsched */ struct sigaction sa; +#ifdef atomic_store_rel_ptr atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv); +#else + *((volatile uintptr_t*) &_glSigArg) = (uintptr_t) srv; +#endif memset(&sa, 0, sizeof sa); sigemptyset(&sa.sa_mask); @@ -1599,14 +1606,17 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) /* if send kill to blob server */ rpc_srv_endBLOBServer(*psrv); + /* wait for BLOB server done */ + while (*(&(*psrv)->srv_blob.root)) + usleep(1000); (*psrv)->srv_kill = 1; sleep(RPC_SCHED_POLLING); - schedEnd(&(*psrv)->srv_root); - if ((*psrv)->srv_server.cli_sa.sa.sa_family == AF_LOCAL) unlink((*psrv)->srv_server.cli_sa.sun.sun_path); + + schedEnd(&(*psrv)->srv_root); pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx); e_free(*psrv);