--- libaitrpc/src/srv.c 2013/11/15 00:21:46 1.21.2.3 +++ libaitrpc/src/srv.c 2014/01/28 14:05:43 1.23 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.21.2.3 2013/11/15 00:21:46 misho Exp $ +* $Id: srv.c,v 1.23 2014/01/28 14:05:43 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, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 +Copyright 2004 - 2014 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -66,7 +66,11 @@ static sched_task_func_t cbProto[SOCK_RAW + 1][4] = { { NULL, NULL, NULL, NULL } /* SOCK_RAW */ }; +/* Global Signal Argument when kqueue support disabled */ +static volatile uintptr_t _glSigArg = 0; + + void rpc_freeCli(rpc_cli_t * __restrict c) { @@ -223,7 +227,7 @@ txPacket(sched_task_t *task) if (ret) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); /* close connection */ schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); @@ -346,7 +350,7 @@ rxPacket(sched_task_t *task) if (rlen) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); return NULL; @@ -448,7 +452,7 @@ txUDPPacket(sched_task_t *task) rpc_func_t *f = NULL; u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc = (struct tagRPCCall*) buf; - int ret, wlen = sizeof(struct tagRPCCall); + int ret, estlen, wlen = sizeof(struct tagRPCCall); struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; struct pollfd pfd; @@ -464,6 +468,13 @@ txUDPPacket(sched_task_t *task) rpc->call_rep.ret = RPC_ERROR(-1); rpc->call_rep.eno = RPC_ERROR(rpc_Errno); } else { + /* calc estimated length */ + estlen = ait_resideVars(RPC_RETVARS(c)) + wlen; + if (estlen > AIT_LEN(&c->cli_buf)) + AIT_RE_BUF(&c->cli_buf, estlen); + buf = AIT_GET_BUF(&c->cli_buf); + rpc = (struct tagRPCCall*) buf; + rpc->call_argc = htons(array_Size(RPC_RETVARS(c))); /* Go Encapsulate variables */ ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, @@ -495,7 +506,7 @@ txUDPPacket(sched_task_t *task) if (ret) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); /* close connection */ schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); @@ -568,7 +579,7 @@ rxUDPPacket(sched_task_t *task) if (rlen) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c, 0, NULL, 0); return NULL; @@ -753,7 +764,8 @@ end: static void * flushBLOB(sched_task_t *task) { - rpc_srv_t *srv = TASK_ARG(task); + uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg); + rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task); rpc_blob_t *b, *tmp; TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) { @@ -763,7 +775,17 @@ flushBLOB(sched_task_t *task) e_free(b); } - schedSignalSelf(task); + if (!schedSignalSelf(task)) { + /* disabled kqueue support in libaitsched */ + struct sigaction sa; + + memset(&sa, 0, sizeof sa); + sigemptyset(&sa.sa_mask); + sa.sa_handler = (void (*)(int)) flushBLOB; + sa.sa_flags = SA_RESTART | SA_RESETHAND; + sigaction(SIGFBLOB, &sa, NULL); + } + return NULL; } @@ -977,7 +999,19 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) return -1; } - schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0); + if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) { + /* disabled kqueue support in libaitsched */ + struct sigaction sa; + + atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv); + + memset(&sa, 0, sizeof sa); + sigemptyset(&sa.sa_mask); + sa.sa_handler = (void (*)(int)) flushBLOB; + sa.sa_flags = SA_RESTART | SA_RESETHAND; + sigaction(SIGFBLOB, &sa, NULL); + } + if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv, srv->srv_blob.server.cli_sock, NULL, 0)) { rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());