[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.8.2.2 and 1.9.2.1

version 1.8.2.2, 2012/05/11 08:46:41 version 1.9.2.1, 2012/05/14 14:51:57
Line 60  txPacket(sched_task_t *task) Line 60  txPacket(sched_task_t *task)
         int ret, wlen = sizeof(struct tagRPCCall);          int ret, wlen = sizeof(struct tagRPCCall);
         array_t *arr = NULL;          array_t *arr = NULL;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (rpc->call_argc) {          if (rpc->call_argc) {
                 f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));                  f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
                 if (!f) {                  if (!f) {
Line 96  txPacket(sched_task_t *task) Line 94  txPacket(sched_task_t *task)
         else if (ret != wlen)          else if (ret != wlen)
                 rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "                  rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                                 "really sended %d bytes", wlen, ret);                                  "really sended %d bytes", wlen, ret);
         else  
                 ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", ret);  
   
         return NULL;          return NULL;
 }  }
Line 113  execCall(sched_task_t *task) Line 109  execCall(sched_task_t *task)
         struct tagRPCCall *rpc = (struct tagRPCCall*) buf;          struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
         int argc = ntohs(rpc->call_argc);          int argc = ntohs(rpc->call_argc);
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         /* Go decapsulate variables ... */          /* Go decapsulate variables ... */
         if (argc) {          if (argc) {
                 arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),                   arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), 
Line 135  execCall(sched_task_t *task) Line 129  execCall(sched_task_t *task)
                 rpc->call_rep.ret = RPC_ERROR(-1);                  rpc->call_rep.ret = RPC_ERROR(-1);
                 rpc->call_rep.eno = RPC_ERROR(rpc_Errno);                  rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
         } else {          } else {
                 ioDEBUG(RPC_DEBUG_LEVEL, "RPC function %s from module %s",   
                                 AIT_GET_STR(&f->func_name), AIT_GET_LIKE(&f->func_file, char*));  
   
                 /* if client doesn't want reply */                  /* if client doesn't want reply */
                 argc = rpc->call_req.flags & RPC_NOREPLY;                  argc = rpc->call_req.flags & RPC_NOREPLY;
                 rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));                  rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
Line 171  rxPacket(sched_task_t *task) Line 162  rxPacket(sched_task_t *task)
         struct tagRPCCall *rpc;          struct tagRPCCall *rpc;
         struct timespec ts;          struct timespec ts;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         memset(buf, 0, TASK_DATLEN(task));          memset(buf, 0, TASK_DATLEN(task));
         rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);          rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
         if (rlen == -1) {          if (rlen == -1) {
                 LOGERR;                  LOGERR;
                s->srv_kill = s->srv_blob.state = kill;                c->cli_kill = kill;
                 return NULL;                  return NULL;
         } else if (!rlen) {     /* receive EOF */          } else if (!rlen) {     /* receive EOF */
                s->srv_kill = s->srv_blob.state = kill;                c->cli_kill = kill;
                 return NULL;                  return NULL;
        } else        }
                ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen); 
   
         do {          do {
                 if (rlen < sizeof(struct tagRPCCall)) {                  if (rlen < sizeof(struct tagRPCCall)) {
Line 252  rpc_srv_dispatchCall(void *arg) Line 240  rpc_srv_dispatchCall(void *arg)
         sched_root_task_t *root;          sched_root_task_t *root;
         struct timespec ts = { DEF_RPC_TIMEOUT, 0 };          struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!arg) {          if (!arg) {
                 rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");                  rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
                 return NULL;                  return NULL;
Line 279  rpc_srv_dispatchCall(void *arg) Line 265  rpc_srv_dispatchCall(void *arg)
   
         schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);          schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);
   
        schedRun(root, (void*) &s->srv_kill);        schedRun(root, (void*) &c->cli_kill);
         schedEnd(&root);          schedEnd(&root);
   
         shutdown(c->cli_sock, SHUT_RDWR);          shutdown(c->cli_sock, SHUT_RDWR);
Line 297  txBLOB(sched_task_t *task) Line 283  txBLOB(sched_task_t *task)
         struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;          struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
         int wlen = sizeof(struct tagBLOBHdr);          int wlen = sizeof(struct tagBLOBHdr);
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         /* calculate CRC */          /* calculate CRC */
         blob->hdr_crc ^= blob->hdr_crc;          blob->hdr_crc ^= blob->hdr_crc;
         blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));          blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
Line 310  txBLOB(sched_task_t *task) Line 294  txBLOB(sched_task_t *task)
         else if (wlen != sizeof(struct tagBLOBHdr))          else if (wlen != sizeof(struct tagBLOBHdr))
                 rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "                  rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                                 "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);                                  "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
         else  
                 ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", wlen);  
   
         return NULL;          return NULL;
 }  }
Line 328  rxBLOB(sched_task_t *task) Line 310  rxBLOB(sched_task_t *task)
         u_short crc;          u_short crc;
         struct timespec ts;          struct timespec ts;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         /* check for disable service at this moment? */          /* check for disable service at this moment? */
         if (!s || s->srv_blob.state == disable) {          if (!s || s->srv_blob.state == disable) {
                 usleep(100000);                  usleep(100000);
Line 345  rxBLOB(sched_task_t *task) Line 325  rxBLOB(sched_task_t *task)
         rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);          rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
         if (rlen == -1) {          if (rlen == -1) {
                 LOGERR;                  LOGERR;
                s->srv_blob.state = kill;                c->cli_kill = kill;
                 return NULL;                  return NULL;
        } else if (!rlen || s->srv_kill == kill) { /* receive EOF */        } else if (!rlen) { /* receive EOF */
                s->srv_blob.state = kill;                c->cli_kill = kill;
                 return NULL;                  return NULL;
        } else        }
                ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen); 
   
         if (rlen < sizeof(struct tagBLOBHdr)) {          if (rlen < sizeof(struct tagBLOBHdr)) {
                 rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");                  rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
Line 444  rpc_srv_dispatchVars(void *arg) Line 423  rpc_srv_dispatchVars(void *arg)
         u_char *buf;          u_char *buf;
         struct timespec ts = { DEF_RPC_TIMEOUT, 0 };          struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!arg) {          if (!arg) {
                 rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");                  rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
                 return NULL;                  return NULL;
Line 471  rpc_srv_dispatchVars(void *arg) Line 448  rpc_srv_dispatchVars(void *arg)
   
         schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));          schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
   
        schedRun(root, (void*) &s->srv_blob.state);        schedRun(root, (void*) &c->cli_kill);
         schedEnd(&root);          schedEnd(&root);
   
         shutdown(c->cli_sock, SHUT_RDWR);          shutdown(c->cli_sock, SHUT_RDWR);
Line 497  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 474  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
         int n = 1;          int n = 1;
         io_sockaddr_t sa;          io_sockaddr_t sa;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!srv) {          if (!srv) {
                 rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");                  rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
                 return -1;                  return -1;
Line 579  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 554  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
         } 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));
   
         pthread_mutex_init(&srv->srv_blob.mtx, NULL);  
   
         rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);          rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
         rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);          rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);
         rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);          rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);
Line 603  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 576  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
         register int i;          register int i;
         rpc_blob_t *f;          rpc_blob_t *f;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!srv || srv->srv_blob.state == disable)          if (!srv || srv->srv_blob.state == disable)
                 return;                  return;
         else          else
Line 621  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 592  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
                         shutdown(c->cli_sock, SHUT_RDWR);                          shutdown(c->cli_sock, SHUT_RDWR);
         close(srv->srv_blob.server.cli_sock);          close(srv->srv_blob.server.cli_sock);
   
         pthread_mutex_lock(&srv->srv_blob.mtx);  
         if (srv->srv_blob.clients) {          if (srv->srv_blob.clients) {
                 free(srv->srv_blob.clients);                  free(srv->srv_blob.clients);
                 srv->srv_blob.clients = NULL;                  srv->srv_blob.clients = NULL;
Line 633  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 603  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
                 rpc_srv_blobFree(srv, f);                  rpc_srv_blobFree(srv, f);
                 free(f);                  free(f);
         }          }
         pthread_mutex_unlock(&srv->srv_blob.mtx);  
   
         AIT_FREE_VAL(&srv->srv_blob.dir);          AIT_FREE_VAL(&srv->srv_blob.dir);
   
         while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);  
         pthread_mutex_destroy(&srv->srv_blob.mtx);  
   
         /* at final, save disable BLOB service state */          /* at final, save disable BLOB service state */
         srv->srv_blob.state = disable;          srv->srv_blob.state = disable;
 }  }
Line 660  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 626  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
         pthread_attr_t attr;          pthread_attr_t attr;
         struct pollfd pfd;          struct pollfd pfd;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!srv || srv->srv_blob.state == kill) {          if (!srv || srv->srv_blob.state == kill) {
                 rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");                  rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
                 return -1;                  return -1;
Line 671  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 635  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
                 LOGERR;                  LOGERR;
                 return -1;                  return -1;
         } else          } else
                fcntl(srv->srv_server.cli_sock, F_SETFL, fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);                fcntl(srv->srv_blob.server.cli_sock, F_SETFL, 
                                 fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
   
         pthread_attr_init(&attr);          pthread_attr_init(&attr);
         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);          pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
   
        pfd.fd = srv->srv_server.cli_sock;        pfd.fd = srv->srv_blob.server.cli_sock;
         pfd.events = POLLIN | POLLPRI;          pfd.events = POLLIN | POLLPRI;
         /* main BLOB loop */          /* main BLOB loop */
         while (srv->srv_blob.state != kill && srv->srv_kill != kill) {          while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
Line 686  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 651  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
                 if (i >= srv->srv_numcli) {                  if (i >= srv->srv_numcli) {
 #ifdef HAVE_PTHREAD_YIELD  #ifdef HAVE_PTHREAD_YIELD
                         pthread_yield();                          pthread_yield();
 #else  
                         usleep(1000000);  
 #endif  #endif
                           usleep(1000000);
                         continue;                          continue;
                 }                  }
   
Line 708  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 672  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
                         c->cli_parent = srv;                          c->cli_parent = srv;
   
                 /* spawn dispatch thread for BLOB client */                  /* spawn dispatch thread for BLOB client */
                   c->cli_kill = enable;
                 if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {                  if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
                         LOGERR;                          LOGERR;
                         continue;                          continue;
Line 742  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 707  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         struct hostent *host = NULL;          struct hostent *host = NULL;
         io_sockaddr_t sa;          io_sockaddr_t sa;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!concurentClients || !regProgID ||           if (!concurentClients || !regProgID || 
                         (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {                          (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
                 rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");                  rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
Line 871  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 834  rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
         register int i;          register int i;
         rpc_func_t *f;          rpc_func_t *f;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!psrv || !*psrv) {          if (!psrv || !*psrv) {
                 rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");                  rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
                 return;                  return;
Line 923  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 884  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
         pthread_attr_t attr;          pthread_attr_t attr;
         struct pollfd pfd;          struct pollfd pfd;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         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;
Line 933  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 892  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
         /* 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);
   
        if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {        if (listen(srv->srv_server.cli_sock, srv->srv_numcli) == -1) {
                 LOGERR;                  LOGERR;
                 return -1;                  return -1;
         } else          } else
Line 952  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 911  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
                 if (i >= srv->srv_numcli) {                  if (i >= srv->srv_numcli) {
 #ifdef HAVE_PTHREAD_YIELD  #ifdef HAVE_PTHREAD_YIELD
                         pthread_yield();                          pthread_yield();
 #else  
                         usleep(1000000);  
 #endif  #endif
                           usleep(1000000);
                         continue;                          continue;
                 }                  }
   
Line 974  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 932  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
                         c->cli_parent = srv;                          c->cli_parent = srv;
   
                 /* spawn rpc client dispatcher */                  /* spawn rpc client dispatcher */
                   c->cli_kill = enable;
                 if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {                  if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {
                         LOGERR;                          LOGERR;
                         continue;                          continue;
Line 1001  rpc_srv_execCall(rpc_func_t * __restrict call, struct  Line 960  rpc_srv_execCall(rpc_func_t * __restrict call, struct 
         void *dl;          void *dl;
         rpc_callback_t func;          rpc_callback_t func;
         int ret;          int ret;
   
         ioTRACE(RPC_TRACE_LEVEL);  
   
         if (!call || !rpc || !call->func_parent) {          if (!call || !rpc || !call->func_parent) {
                 rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");                  rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
Removed from v.1.8.2.2  
changed lines
  Added in v.1.9.2.1

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