[BACK]Return to srv.c CVS log [TXT] [DIR] Up to [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src

Diff for /libaitrpc/src/srv.c between versions 1.6.2.1 and 1.6.2.6

version 1.6.2.1, 2012/03/12 14:53:07 version 1.6.2.6, 2012/03/13 17:21:52
Line 46  SUCH DAMAGE. Line 46  SUCH DAMAGE.
 #include "global.h"  #include "global.h"
   
   
   static void *rxPacket(sched_task_t*);
   
 static void *  static void *
rpc_srv_dispatchCall(void *arg)txPacket(sched_task_t *task)
 {  {
        rpc_cli_t *c = arg;        rpc_cli_t *c = TASK_ARG(task);
        rpc_srv_t *s;        rpc_srv_t *s = c->cli_parent;
         rpc_func_t *f = NULL;          rpc_func_t *f = NULL;
        array_t *arr;        u_char *buf = TASK_DATA(task);
         struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
         int ret, wlen = sizeof(struct tagRPCCall);
         array_t *arr = NULL;
 
         if (rpc->call_argc) {
                 f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
                 if (!f) {
                         rpc->call_argc ^= rpc->call_argc;
                         rpc->call_rep.ret = RPC_ERROR(-1);
                         rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                 } else {
                         rpc->call_argc = htons(rpc_srv_getVars(f, &arr));
                         /* Go Encapsulate variables */
                         ret = io_vars2buffer(buf + wlen, TASK_DATLEN(task) - wlen, arr);
                         io_clrVars(f->func_vars);
                         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);
                         } else
                                 wlen += ret;
                 }
         }
 
         /* send reply */
         ret = send(TASK_FD(task), buf, wlen, 0);
         if (ret == -1)
                 LOGERR;
         else if (ret != wlen)
                 rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                                 "really sended %d bytes", wlen, ret);
 
         /* lets get next packet */
         schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                         TASK_DATA(task), TASK_DATLEN(task));
         return NULL;
 }
 
 static void *
 execCall(sched_task_t *task)
 {
         rpc_cli_t *c = TASK_ARG(task);
         rpc_srv_t *s = c->cli_parent;
         rpc_func_t *f = NULL;
         array_t *arr = NULL;
         u_char *buf = TASK_DATA(task);
         struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
         int argc = ntohs(rpc->call_argc);
 
         /* Go decapsulate variables ... */
         if (!(rpc->call_req.flags & RPC_NOREPLY) && argc) {
                 arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), 
                                 TASK_DATLEN(task) - sizeof(struct tagRPCCall), argc, 1);
                 if (!arr) {
                         rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());
                         rpc->call_argc ^= rpc->call_argc;
                         rpc->call_rep.ret = RPC_ERROR(-1);
                         rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                         return NULL;
                 }
         }
 
         if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash)))) {
                 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);
         } else {
                 rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
                 if (rpc->call_rep.ret == htonl(-1)) {
                         rpc->call_rep.eno = RPC_ERROR(errno);
                         rpc->call_argc ^= rpc->call_argc;
                 } else {
                         rpc->call_rep.eno ^= rpc->call_rep.eno;
                         rpc->call_argc = htons(rpc_srv_getVars(f, NULL));
                 }
         }
 
         if (arr)
                 io_arrayDestroy(&arr);
         return NULL;
 }
 
 static void *
 rxPacket(sched_task_t *task)
 {
         rpc_cli_t *c = TASK_ARG(task);
         rpc_srv_t *s = c->cli_parent;
         u_char *buf = TASK_DATA(task);
         int rlen;
         u_short crc;
         struct tagRPCCall *rpc;          struct tagRPCCall *rpc;
        struct tagRPCRet *rrpc;        struct timespec ts;
        rpc_sess_t ses = { 0 }; 
        fd_set fds; 
        u_char *buf; 
        int ret, argc = 0, Limit = 0; 
        register int i; 
        uint16_t tag = 0; 
        uint32_t hash = 0; 
   
        if (!arg) {        memset(buf, 0, TASK_DATLEN(task));
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n");        rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
         if (rlen == -1) {
                 LOGERR;
                 s->srv_kill = kill;
 
                 schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;                  return NULL;
        } else        } else if (!rlen) {    /* receive EOF */
                s = c->cli_parent;                s->srv_kill = kill;
   
        buf = malloc(s->srv_netbuf);                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
        if (!buf) {                                TASK_DATA(task), TASK_DATLEN(task));
                LOGERR; 
                 return NULL;                  return NULL;
         }          }
           if (rlen < sizeof(struct tagRPCCall)) {
                   rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
   
        do {                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                FD_ZERO(&fds);                                TASK_DATA(task), TASK_DATLEN(task));
                FD_SET(c->cli_sock, &fds);                return NULL;
                ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);        } else
                if (ret == -1) {                rpc = (struct tagRPCCall*) buf;
                        if (errno == EINTR && s->srv_kill != kill) 
                                continue; 
   
                        LOGERR;        /* check integrity of packet */
                        ret = -2;        crc = ntohs(rpc->call_crc);
                        break;        rpc->call_crc ^= rpc->call_crc;
                }        if (crc != crcFletcher16((u_short*) buf, ((rlen + 1) & ~1) / 2)) {
                memset(buf, 0, s->srv_netbuf);                rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
                ret = recv(c->cli_sock, buf, s->srv_netbuf, 0); 
                if (ret == -1) { 
                        LOGERR; 
                        ret = -3; 
                        break; 
                } 
                if (!ret) {             /* receive EOF */ 
                        ret = 0; 
                        break; 
                } 
                if (ret < sizeof(struct tagRPCCall)) { 
                        rpc_SetErr(ERPCMISMATCH, "Error:: too short RPC packet ...\n"); 
                        ret = -4; 
                        if (s->srv_kill != kill) 
                                continue; 
                        else 
                                break; 
                } else 
                        rpc = (struct tagRPCCall*) buf; 
                /* check RPC packet session info */ 
                if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) { 
                        rpc_SetErr(ERPCMISMATCH, "Error:: get invalid RPC session ...\n"); 
                        ret = -5; 
                        goto makeReply; 
                } else 
                        Limit = sizeof(struct tagRPCCall); 
   
                tag = rpc->call_tag;                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                hash = rpc->call_hash;                                TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;
         }
   
                /* RPC is OK! Go decapsulate variables ... */        /* check RPC packet session info */
                if (ntohs(rpc->call_argc)) {        if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
                        arr = io_buffer2vars(buf + Limit, s->srv_netbuf - Limit,                 rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                                        ntohs(rpc->call_argc), 1);                rpc->call_argc ^= rpc->call_argc;
                        if (!arr) {                rpc->call_rep.ret = RPC_ERROR(-1);
                                ret = -5;                rpc->call_rep.eno = RPC_ERROR(errno);
                                goto makeReply;                goto end;
                        }        } else {
                } else                /* change socket timeout from last packet */
                        arr = NULL;                ts.tv_sec = rpc->call_session.sess_timeout;
                 ts.tv_nsec = 0;
                 schedPolling(TASK_ROOT(task), &ts, NULL);
         }
   
                /* execute call */        /* execute RPC call */
                argc = 0;        schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), 0, 
                memcpy(&ses, &rpc->call_session, sizeof ses);                        TASK_DATA(task), TASK_DATLEN(task));
                if (!(f = rpc_srv_getCall(s, ntohs(tag), ntohl(hash)))) { 
                        rpc_SetErr(EPROGUNAVAIL, "Error:: call not found into RPC server ...\n"); 
                        ret = -6; 
                } else 
                        if ((ret = rpc_srv_execCall(f, rpc, arr)) == -1) 
                                ret = -9; 
                        else { 
                                if (arr) 
                                        io_arrayDestroy(&arr); 
                                argc = rpc_srv_getVars(f, &arr); 
                                goto makeReply;         /* Call finish OK */ 
                        } 
   
                if (arr)end:
                        io_arrayDestroy(&arr);        /* send RPC reply */
         if (!(rpc->call_req.flags & RPC_NOREPLY))
                 schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
         else
                 schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
         return NULL;
 }
   
makeReply:static void *
                /* Made reply */rpc_srv_dispatchCall(void *arg)
                memset(buf, 0, s->srv_netbuf);{
                rrpc = (struct tagRPCRet*) buf;        rpc_cli_t *c = arg;
                Limit = sizeof(struct tagRPCRet);        rpc_srv_t *s;
         u_char *buf;
         sched_root_task_t *root;
         struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
   
                memcpy(&rrpc->ret_session, &ses, sizeof(rpc_sess_t));        if (!arg) {
                rrpc->ret_tag = tag;                rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
                rrpc->ret_hash = hash;                return NULL;
                rrpc->ret_errno = htonl(rpc_Errno);        } else
                rrpc->ret_retcode = htonl(ret);                s = c->cli_parent;
                rrpc->ret_argc = htons(argc); 
   
                if (argc && arr) {        /* allocate net buffer */
                        /* Go Encapsulate variables ... */        buf = malloc(s->srv_netbuf);
                        if ((i = io_vars2buffer(buf + Limit, s->srv_netbuf - Limit, arr)) == -1) {        if (!buf) {
                                io_clrVars(f->func_vars);                LOGERR;
                                argc = 0;                return NULL;
                                ret = -7;        }
                                rpc_SetErr(EBADRPC, "Error:: in prepare RPC packet values (-7) ...\n"); 
                                goto makeReply; 
                        } else { 
                                Limit += i; 
   
                                io_clrVars(f->func_vars);        root = schedBegin();
                        }        if (!root) {
                }                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                 free(buf);
                 return NULL;
         } else {
                 schedTermCondition(root, kill);
                 schedPolling(root, &ts, NULL);
         }
   
                ret = send(c->cli_sock, buf, Limit, 0);        schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);
                if (ret == -1) { 
                        LOGERR; 
                        ret = -8; 
                        break; 
                } 
                if (ret != Limit) { 
                        rpc_SetErr(EPROCUNAVAIL, "Error:: in send RPC request, should be send %d bytes, " 
                                        "really is %d\n", Limit, ret); 
                        ret = -9; 
                        if (s->srv_kill != kill) 
                                continue; 
                        else 
                                break; 
                } 
        } while (ret > -1 || s->srv_kill != kill); 
   
           schedRun(root, (void*) &s->srv_kill);
           schedEnd(&root);
   
         shutdown(c->cli_sock, SHUT_RDWR);          shutdown(c->cli_sock, SHUT_RDWR);
         close(c->cli_sock);          close(c->cli_sock);
         memset(c, 0, sizeof(rpc_cli_t));          memset(c, 0, sizeof(rpc_cli_t));
         free(buf);          free(buf);
        return (void*) (long)ret;        return NULL;
 }  }
   
   
Line 335  makeReply: Line 394  makeReply:
 // -------------------------------------------------  // -------------------------------------------------
   
 /*  /*
 * rpc_srv_initBLOBServer() Init & create BLOB Server * rpc_srv_initBLOBServer() - Init & create BLOB Server
  *
  * @srv = RPC server instance   * @srv = RPC server instance
  * @Port = Port for bind server, if Port == 0 default port is selected   * @Port = Port for bind server, if Port == 0 default port is selected
  * @diskDir = Disk place for BLOB file objects   * @diskDir = Disk place for BLOB file objects
Line 361  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 421  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
                 LOGERR;                  LOGERR;
                 return -1;                  return -1;
         } else          } else
                srv->srv_blob.dir = strdup(diskDir);                AIT_SET_STR(&srv->srv_blob.dir, diskDir);
   
         srv->srv_blob.server.cli_tid = pthread_self();          srv->srv_blob.server.cli_tid = pthread_self();
         srv->srv_blob.server.cli_parent = srv;          srv->srv_blob.server.cli_parent = srv;
Line 378  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 438  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
                         strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);                          strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);
                         break;                          break;
                 default:                  default:
                        free(srv->srv_blob.dir);                        AIT_FREE_VAL(&srv->srv_blob.dir);
                         return -1;                          return -1;
         }          }
         memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);          memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);
Line 387  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 447  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
         srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);          srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
         if (srv->srv_blob.server.cli_sock == -1) {          if (srv->srv_blob.server.cli_sock == -1) {
                 LOGERR;                  LOGERR;
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         }          }
         if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_blob.server.cli_sock);                  close(srv->srv_blob.server.cli_sock);
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         }          }
         n = srv->srv_netbuf;          n = srv->srv_netbuf;
         if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_blob.server.cli_sock);                  close(srv->srv_blob.server.cli_sock);
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         }          }
         if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_blob.server.cli_sock);                  close(srv->srv_blob.server.cli_sock);
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         }          }
         if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,           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) {                                  srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_blob.server.cli_sock);                  close(srv->srv_blob.server.cli_sock);
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         }          }
   
Line 422  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 482  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
         if (!srv->srv_blob.clients) {          if (!srv->srv_blob.clients) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_blob.server.cli_sock);                  close(srv->srv_blob.server.cli_sock);
                free(srv->srv_blob.dir);                AIT_FREE_VAL(&srv->srv_blob.dir);
                 return -1;                  return -1;
         } else          } else
                 memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));                  memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
Line 439  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 499  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
 }  }
   
 /*  /*
 * rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
  *
  * @srv = RPC Server instance   * @srv = RPC Server instance
  * return: none   * return: none
  */   */
Line 461  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 522  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
         rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);          rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
         rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);          rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
   
        if (srv->srv_blob.dir)        AIT_FREE_VAL(&srv->srv_blob.dir);
                free(srv->srv_blob.dir); 
   
         /* close all clients connections & server socket */          /* close all clients connections & server socket */
         for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)          for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
Line 489  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 549  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
 }  }
   
 /*  /*
 * rpc_srv_loopBLOB() Execute Main BLOB server loop and wait for clients requests * rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
  *
  * @srv = RPC Server instance   * @srv = RPC Server instance
  * return: -1 error or 0 ok, infinite loop ...   * return: -1 error or 0 ok, infinite loop ...
  */   */
Line 566  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 627  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
   
   
 /*  /*
 * rpc_srv_initServer() Init & create RPC Server * rpc_srv_initServer() - Init & create RPC Server
  *
  * @regProgID = ProgramID for authentication & recognition   * @regProgID = ProgramID for authentication & recognition
  * @regProcID = ProcessID for authentication & recognition   * @regProcID = ProcessID for authentication & recognition
  * @concurentClients = Concurent clients at same time to this server   * @concurentClients = Concurent clients at same time to this server
Line 645  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 707  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         srv->srv_server.cli_parent = srv;          srv->srv_server.cli_parent = srv;
         memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);          memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);
   
         /* init rpc scheduler */  
         srv->srv_sched = schedBegin();  
         if (!srv->srv_sched) {  
                 rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());  
                 return NULL;  
         }  
   
         /* create server socket */          /* create server socket */
         srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);          srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
         if (srv->srv_server.cli_sock == -1) {          if (srv->srv_server.cli_sock == -1) {
                 LOGERR;                  LOGERR;
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         }          }
         if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_server.cli_sock);                  close(srv->srv_server.cli_sock);
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         }          }
Line 671  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 724  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_server.cli_sock);                  close(srv->srv_server.cli_sock);
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         }          }
         if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {          if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_server.cli_sock);                  close(srv->srv_server.cli_sock);
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         }          }
Line 686  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 737  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
                                 srv->srv_server.cli_sa.sa.sa_len) == -1) {                                  srv->srv_server.cli_sa.sa.sa_len) == -1) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_server.cli_sock);                  close(srv->srv_server.cli_sock);
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         }          }
Line 696  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 746  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         if (!srv->srv_clients) {          if (!srv->srv_clients) {
                 LOGERR;                  LOGERR;
                 close(srv->srv_server.cli_sock);                  close(srv->srv_server.cli_sock);
                 schedEnd(&srv->srv_sched);  
                 free(srv);                  free(srv);
                 return NULL;                  return NULL;
         } else          } else
Line 712  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 761  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
 }  }
   
 /*  /*
 * rpc_srv_endServer() Destroy RPC server, close all opened sockets and free resources * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
  *
  * @psrv = RPC Server instance   * @psrv = RPC Server instance
  * return: none   * return: none
  */   */
Line 730  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 780  rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
   
         rpc_srv_endBLOBServer(*psrv);          rpc_srv_endBLOBServer(*psrv);
   
         /* destroy rpc scheduler */  
         schedEnd(&(*psrv)->srv_sched);  
   
         /* close all clients connections & server socket */          /* close all clients connections & server socket */
         for (i = 0, c = (*psrv)->srv_clients; i < (*psrv)->srv_numcli && c; i++, c++)          for (i = 0, c = (*psrv)->srv_clients; i < (*psrv)->srv_numcli && c; i++, c++)
                 if (c->cli_sa.sa.sa_family) {                  if (c->cli_sa.sa.sa_family) {
Line 752  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 799  rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
         while ((f = (*psrv)->srv_funcs)) {          while ((f = (*psrv)->srv_funcs)) {
                 (*psrv)->srv_funcs = f->func_next;                  (*psrv)->srv_funcs = f->func_next;
                 io_freeVars(&f->func_vars);                  io_freeVars(&f->func_vars);
                   AIT_FREE_VAL(&f->func_name);
                   AIT_FREE_VAL(&f->func_file);
                 free(f);                  free(f);
         }          }
         pthread_mutex_unlock(&(*psrv)->srv_mtx);          pthread_mutex_unlock(&(*psrv)->srv_mtx);
Line 764  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 813  rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
 }  }
   
 /*  /*
 * rpc_srv_loopServer() Execute Main server loop and wait for clients requests * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
  *
  * @srv = RPC Server instance   * @srv = RPC Server instance
  * return: -1 error or 0 ok, infinite loop ...   * return: -1 error or 0 ok, infinite loop ...
  */   */
Line 777  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 827  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
         fd_set fds;          fd_set fds;
         int ret;          int ret;
         struct timeval tv = { DEF_RPC_TIMEOUT, 0 };          struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
           pthread_attr_t attr;
   
         if (!srv) {          if (!srv) {
                 rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");                  rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
                 return -1;                  return -1;
         }          }
   
           tv.tv_sec = srv->srv_session.sess_timeout;
   
         /* activate BLOB server worker if srv->srv_blob.state == enable */          /* activate BLOB server worker if srv->srv_blob.state == enable */
         rpc_srv_execBLOBServer(srv);          rpc_srv_execBLOBServer(srv);
   
Line 791  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 844  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
                 return -1;                  return -1;
         }          }
   
           pthread_attr_init(&attr);
           pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
   
           /* main rpc loop */
         while (srv->srv_kill != kill) {          while (srv->srv_kill != kill) {
                 for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)                  for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
                         if (!c->cli_sa.sa.sa_family)                          if (!c->cli_sa.sa.sa_family)
Line 822  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 879  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
                 } else                  } else
                         c->cli_parent = srv;                          c->cli_parent = srv;
   
                if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) {                /* spawn rpc client dispatcher */
                 if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {
                         LOGERR;                          LOGERR;
                         continue;                          continue;
                } else                }
                        pthread_detach(c->cli_tid); 
         }          }
   
           pthread_attr_destroy(&attr);
         return 0;          return 0;
 }  }
   
Line 836  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 894  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
   
 /*  /*
  * rpc_srv_execCall() Execute registered call from RPC server   * rpc_srv_execCall() Execute registered call from RPC server
    *
  * @call = Register RPC call   * @call = Register RPC call
  * @rpc = IN RPC call structure   * @rpc = IN RPC call structure
  * @args = IN RPC calling arguments from RPC client   * @args = IN RPC calling arguments from RPC client
Line 850  rpc_srv_execCall(rpc_func_t * __restrict call, struct  Line 909  rpc_srv_execCall(rpc_func_t * __restrict call, struct 
         int ret;          int ret;
   
         if (!call || !rpc || !call->func_parent) {          if (!call || !rpc || !call->func_parent) {
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t exec call from RPC server ...\n");                rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
                 return -1;                  return -1;
         }          }
   
        dl = dlopen((char*) (*call->func_file ? call->func_file : NULL), RTLD_NOW);        dl = dlopen(AIT_VOID(&call->func_file), RTLD_NOW);
         if (!dl) {          if (!dl) {
                rpc_SetErr(ENOENT, "Error:: Can`t attach module %s!\n", dlerror());                rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
                 return -1;                  return -1;
         }          }
   
        func = dlsym(dl, (char*) call->func_name);        func = dlsym(dl, (const char*) AIT_GET_STR(&call->func_name));
         if (func)          if (func)
                 ret = func(call, ntohs(rpc->call_argc), args);                  ret = func(call, ntohs(rpc->call_argc), args);
         else {          else {
                rpc_SetErr(ENOEXEC, "Error:: Can`t find function %s!\n", dlerror());                rpc_SetErr(ENOEXEC, "Can`t find function %s!", dlerror());
                 ret = -1;                  ret = -1;
         }          }
   
Removed from v.1.6.2.1  
changed lines
  Added in v.1.6.2.6

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>