--- libaitrpc/src/srv.c 2015/06/29 22:39:33 1.26.2.9 +++ libaitrpc/src/srv.c 2024/02/26 11:01:01 1.30.2.4 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.26.2.9 2015/06/29 22:39:33 misho Exp $ +* $Id: srv.c,v 1.30.2.4 2024/02/26 11:01:01 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004 - 2015 +Copyright 2004 - 2024 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -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); @@ -313,7 +312,7 @@ rxPacket(sched_task_t *task) u_short crc; #endif u_char *buf = AIT_GET_BUF(&c->cli_buf); - struct tagRPCCall *rpc = (struct tagRPCCall*) buf; + struct tagRPCCall b, *rpc = (struct tagRPCCall*) buf; #ifdef TCP_SESSION_TIMEOUT struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; @@ -322,6 +321,11 @@ rxPacket(sched_task_t *task) TASK_ARG(task), ts, TASK_ARG(task), 0); #endif + /* prepare rx */ + len = recv(TASK_FD(task), &b, sizeof b, MSG_PEEK); + if (len == sizeof b) + rlen = ntohl(b.call_len); + rlen = recv(TASK_FD(task), buf, rlen, 0); if (rlen == -1) { /* close connection */ @@ -333,8 +337,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); @@ -342,8 +346,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; } @@ -361,8 +365,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 @@ -372,8 +376,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; } @@ -383,7 +387,7 @@ rxPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len); err: /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET], TASK_ARG(task), TASK_FD(task), rpc, len); @@ -397,7 +401,7 @@ acceptClients(sched_task_t *task) { rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; - socklen_t salen = sizeof(sockaddr_t); + socklen_t salen = E_SOCKADDR_MAX; int sock; #ifdef TCP_SESSION_TIMEOUT struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; @@ -420,8 +424,10 @@ acceptClients(sched_task_t *task) AIT_FREE_VAL(&c->cli_buf); array_Del(srv->srv_clients, c->cli_id, 42); goto end; - } else + } else { fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK); + fcntl(c->cli_sock, F_SETFD, FD_CLOEXEC); + } #ifdef TCP_SESSION_TIMEOUT /* armed timer for close stateless connection */ @@ -458,14 +464,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 */ @@ -475,8 +481,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; } @@ -494,7 +500,7 @@ txUDPPacket(sched_task_t *task) /* send reply */ ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL, - &c->cli_sa.sa, c->cli_sa.sa.sa_len); + &c->cli_sa.sa, e_addrlen(&c->cli_sa)); if (ret == -1) { /* close connection */ schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], @@ -513,13 +519,15 @@ rxUDPPacket(sched_task_t *task) u_short crc; struct tagRPCCall *rpc; sockaddr_t sa; - socklen_t salen; + socklen_t salen = E_SOCKADDR_MAX; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; ait_val_t b = AIT_VAL_INIT; /* receive connect packet */ AIT_SET_BUF(&b, NULL, srv->srv_netbuf); - salen = sa.ss.ss_len = sizeof(sockaddr_t); +#ifndef __linux__ + sa.ss.ss_len = salen; +#endif 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)) @@ -536,7 +544,7 @@ rxUDPPacket(sched_task_t *task) /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) rpc, len / 2)) + if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), len / 2)) goto end; /* check RPC packet session info */ @@ -567,7 +575,7 @@ rxUDPPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -598,14 +606,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 */ @@ -613,9 +621,10 @@ txRAWPacket(sched_task_t *task) RPC_RETVARS(c)); if (ret == -1) { 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; } @@ -633,7 +642,7 @@ txRAWPacket(sched_task_t *task) /* send reply */ ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL, - &c->cli_sa.sa, c->cli_sa.sa.sa_len); + &c->cli_sa.sa, e_addrlen(&c->cli_sa)); if (ret == -1) { /* close connection */ schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], @@ -652,13 +661,15 @@ rxRAWPacket(sched_task_t *task) u_short crc; struct tagRPCCall *rpc; sockaddr_t sa; - socklen_t salen; + socklen_t salen = E_SOCKADDR_MAX; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; ait_val_t b = AIT_VAL_INIT; /* receive connect packet */ AIT_SET_BUF(&b, NULL, srv->srv_netbuf); - salen = sa.ss.ss_len = sizeof(sockaddr_t); +#ifndef __linux__ + sa.ss.ss_len = salen; +#endif 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; @@ -670,15 +681,18 @@ rxRAWPacket(sched_task_t *task) rpc = (struct tagRPCCall*) (h + 1); } } else { +#ifdef IPV6_REMOVE_HEADER struct ip6_hdr *h; h = (struct ip6_hdr*) AIT_GET_BUF(&b); - if (rlen < (ntohs(h->ip6_plen) + sizeof(struct ip6_hdr)) || - h->ip6_nxt != IPPROTO_ERPC) + if (rlen < ntohs(h->ip6_plen) || h->ip6_nxt != IPPROTO_ERPC) goto end; else { rlen -= sizeof(struct ip6_hdr); rpc = (struct tagRPCCall*) (h + 1); } +#else + rpc = (struct tagRPCCall*) AIT_GET_BUF(&b); +#endif } if (rlen < sizeof(struct tagRPCCall)) goto end; @@ -694,7 +708,7 @@ rxRAWPacket(sched_task_t *task) /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) rpc, len / 2)) + if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), len / 2)) goto end; /* check RPC packet session info */ @@ -725,7 +739,7 @@ rxRAWPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -738,6 +752,7 @@ end: static void * txBPFPacket(sched_task_t *task) { +#ifndef __linux__ rpc_cli_t *c = TASK_ARG(task); rpc_srv_t *s = c->cli_parent; rpc_func_t *f = NULL; @@ -758,14 +773,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 */ @@ -775,8 +790,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; } @@ -806,6 +821,9 @@ txBPFPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); } +#else + rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!"); +#endif taskExit(task, NULL); } @@ -813,6 +831,7 @@ txBPFPacket(sched_task_t *task) static void * rxBPFPacket(sched_task_t *task) { +#ifndef __linux__ rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; int len, rlen, noreply; @@ -890,12 +909,16 @@ rxBPFPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: AIT_FREE_VAL(&b); schedReadSelf(task); +#else + rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!"); +#endif + taskExit(task, NULL); } @@ -921,14 +944,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 */ @@ -938,8 +961,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; } @@ -1016,7 +1039,7 @@ rxEXTPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -1111,7 +1134,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); @@ -1122,7 +1145,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: @@ -1136,19 +1159,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: @@ -1161,7 +1184,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; @@ -1172,7 +1199,7 @@ flushBLOB(sched_task_t *task) e_free(b); } - if (!schedSignalSelf(task)) { + if (sigArg) { /* disabled kqueue support in libaitsched */ struct sigaction sa; @@ -1181,9 +1208,11 @@ flushBLOB(sched_task_t *task) sa.sa_handler = (void (*)(int)) flushBLOB; sa.sa_flags = SA_RESTART | SA_RESETHAND; sigaction(SIGFBLOB, &sa, NULL); + return NULL; + } else { + schedSignalSelf(task); + taskExit(task, NULL); } - - taskExit(task, NULL); } static void * @@ -1192,7 +1221,7 @@ acceptBLOBClients(sched_task_t *task) rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; register int i; - socklen_t salen = sizeof(sockaddr_t); + socklen_t salen = E_SOCKADDR_MAX; int sock; #ifdef TCP_NOPUSH int n = 1; @@ -1237,6 +1266,7 @@ acceptBLOBClients(sched_task_t *task) setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n); #endif fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK); + fcntl(c->cli_sock, F_SETFD, FD_CLOEXEC); } schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0); @@ -1259,6 +1289,7 @@ int rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir) { int n = 1; + socklen_t salen; if (!srv || srv->srv_kill) { rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server"); @@ -1277,19 +1308,22 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s srv->srv_blob.server.cli_parent = srv; - memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t)); + memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof srv->srv_blob.server.cli_sa); switch (srv->srv_blob.server.cli_sa.sa.sa_family) { case AF_INET: srv->srv_blob.server.cli_sa.sin.sin_port = htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1); + salen = sizeof srv->srv_blob.server.cli_sa.sin; break; case AF_INET6: srv->srv_blob.server.cli_sa.sin6.sin6_port = htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1); + salen = sizeof srv->srv_blob.server.cli_sa.sin6; break; case AF_LOCAL: strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob", sizeof srv->srv_blob.server.cli_sa.sun.sun_path); + salen = sizeof srv->srv_blob.server.cli_sa.sun; break; default: AIT_FREE_VAL(&srv->srv_blob.dir); @@ -1322,8 +1356,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s AIT_FREE_VAL(&srv->srv_blob.dir); return -1; } - if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, - srv->srv_blob.server.cli_sa.sa.sa_len) == -1) { + if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, salen) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); AIT_FREE_VAL(&srv->srv_blob.dir); @@ -1350,7 +1383,8 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s close(srv->srv_blob.server.cli_sock); AIT_FREE_VAL(&srv->srv_blob.dir); return -1; - } + } else + schedSignalDispatch(srv->srv_blob.root, 42); return 0; } @@ -1369,10 +1403,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); } /* @@ -1403,7 +1437,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); @@ -1421,6 +1459,7 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) schedPolling(srv->srv_blob.root, &ts, NULL); /* main rpc loop */ schedRun(srv->srv_blob.root, &srv->srv_blob.kill); + schedSignalDispatch(srv->srv_blob.root, 0); /* detach blobs */ TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) { @@ -1469,6 +1508,7 @@ rpc_srv_initServer(u_char InstID, int concurentClients int n = 1; rpc_srv_t *srv = NULL; sockaddr_t sa = E_SOCKADDR_INIT; + socklen_t salen; if (!concurentClients || (proto < 0 || proto > SOCK_RAW)) { rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server"); @@ -1476,7 +1516,7 @@ rpc_srv_initServer(u_char InstID, int concurentClients } if (!Port && proto < SOCK_RAW) Port = RPC_DEFPORT; - if (!e_gethostbyname(csHost, Port, &sa)) + if (!(salen = e_gethostbyname(csHost, Port, &sa))) return NULL; if (!proto) proto = SOCK_STREAM; @@ -1520,7 +1560,8 @@ rpc_srv_initServer(u_char InstID, int concurentClients pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(concurentClients); @@ -1558,8 +1599,7 @@ rpc_srv_initServer(u_char InstID, int concurentClients LOGERR; goto err; } - if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, - srv->srv_server.cli_sa.sa.sa_len) == -1) { + if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, salen) == -1) { LOGERR; goto err; } else @@ -1592,15 +1632,18 @@ 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); *psrv = NULL; @@ -1640,6 +1683,7 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) schedPolling(srv->srv_root, &ts, NULL); /* main rpc loop */ schedRun(srv->srv_root, &srv->srv_kill); + schedSignalDispatch(srv->srv_root, 0); /* close all clients connections & server socket */ for (i = 0; i < array_Size(srv->srv_clients); i++) { @@ -1713,6 +1757,7 @@ rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta rpc_srv_t * rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface) { +#ifndef __linux__ int n = 1; rpc_srv_t *srv = NULL; sockaddr_t sa = E_SOCKADDR_INIT; @@ -1777,7 +1822,8 @@ rpc_srv_initServer2(u_char InstID, int concurentClient pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(concurentClients); @@ -1839,6 +1885,10 @@ err: /* error condition */ schedEnd(&srv->srv_root); pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); +#else + rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!"); +#endif + return NULL; } @@ -1891,7 +1941,8 @@ rpc_srv_initServerExt(u_char InstID, int netBuf, int f pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(1);