[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.6 and 1.6.2.8

version 1.6.2.6, 2012/03/13 17:21:52 version 1.6.2.8, 2012/03/14 15:08:03
Line 47  SUCH DAMAGE. Line 47  SUCH DAMAGE.
   
   
 static void *rxPacket(sched_task_t*);  static void *rxPacket(sched_task_t*);
   static void *rxBLOB(sched_task_t*);
   
 static void *  static void *
 txPacket(sched_task_t *task)  txPacket(sched_task_t *task)
Line 59  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;
   
           FTRACE();
   
         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 80  txPacket(sched_task_t *task) Line 83  txPacket(sched_task_t *task)
                 }                  }
         }          }
   
           /* calculate CRC */
           rpc->call_crc ^= rpc->call_crc;
           rpc->call_crc = htons(crcFletcher16((u_short*) buf, ((wlen + 1) & ~1) / 2));
   
         /* send reply */          /* send reply */
         ret = send(TASK_FD(task), buf, wlen, 0);          ret = send(TASK_FD(task), buf, wlen, 0);
         if (ret == -1)          if (ret == -1)
Line 87  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
                   LOGGER("Sended %d bytes", ret);
   
         /* lets get next packet */  
         schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),   
                         TASK_DATA(task), TASK_DATLEN(task));  
         return NULL;          return NULL;
 }  }
   
Line 105  execCall(sched_task_t *task) Line 111  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);
   
           FTRACE();
   
         /* Go decapsulate variables ... */          /* Go decapsulate variables ... */
         if (!(rpc->call_req.flags & RPC_NOREPLY) && argc) {          if (!(rpc->call_req.flags & RPC_NOREPLY) && argc) {
                 arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),                   arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), 
Line 124  execCall(sched_task_t *task) Line 132  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 {
                   LOGGER("RPC function %s from module %s", AIT_GET_STR(&f->func_name), 
                                   AIT_GET_LIKE(&f->func_file, char*));
   
                 rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));                  rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
                 if (rpc->call_rep.ret == htonl(-1)) {                  if (rpc->call_rep.ret == htonl(-1)) {
                         rpc->call_rep.eno = RPC_ERROR(errno);                          rpc->call_rep.eno = RPC_ERROR(errno);
Line 150  rxPacket(sched_task_t *task) Line 161  rxPacket(sched_task_t *task)
         struct tagRPCCall *rpc;          struct tagRPCCall *rpc;
         struct timespec ts;          struct timespec ts;
   
           FTRACE();
   
         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 = kill;                s->srv_kill = s->srv_blob.state = kill;
 
                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),  
                                TASK_DATA(task), TASK_DATLEN(task)); 
                 return NULL;                  return NULL;
         } else if (!rlen) {     /* receive EOF */          } else if (!rlen) {     /* receive EOF */
                s->srv_kill = kill;                s->srv_kill = s->srv_blob.state = kill;
 
                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),  
                                TASK_DATA(task), TASK_DATLEN(task)); 
                 return NULL;                  return NULL;
        }        } else
                 LOGGER("Readed %d bytes", rlen);
 
         if (rlen < sizeof(struct tagRPCCall)) {          if (rlen < sizeof(struct tagRPCCall)) {
                 rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");                  rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
   
Line 209  end: Line 218  end:
         if (!(rpc->call_req.flags & RPC_NOREPLY))          if (!(rpc->call_req.flags & RPC_NOREPLY))
                 schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task),                   schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));                                  TASK_DATA(task), TASK_DATLEN(task));
        else        /* lets get next packet */
                schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),         schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                                TASK_DATA(task), TASK_DATLEN(task));                        TASK_DATA(task), TASK_DATLEN(task));
         return NULL;          return NULL;
 }  }
   
Line 224  rpc_srv_dispatchCall(void *arg) Line 233  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 };
   
           FTRACE();
   
         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 261  rpc_srv_dispatchCall(void *arg) Line 272  rpc_srv_dispatchCall(void *arg)
   
   
 static void *  static void *
rpc_srv_dispatchVars(void *arg)txBLOB(sched_task_t *task)
 {  {
        rpc_cli_t *c = arg;        rpc_cli_t *c = TASK_ARG(task);
        rpc_srv_t *s;        u_char *buf = TASK_DATA(task);
         struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
         int wlen = sizeof(struct tagBLOBHdr);
 
         FTRACE();
 
         /* calculate CRC */
         blob->hdr_crc ^= blob->hdr_crc;
         blob->hdr_crc = htons(crcFletcher16((u_short*) buf, ((wlen + 1) & ~1) / 2));
 
         /* send reply */
         wlen = send(TASK_FD(task), buf, wlen, 0);
         if (wlen == -1)
                 LOGERR;
         else if (wlen != sizeof(struct tagBLOBHdr))
                 rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                                 "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
         else
                 LOGGER("Sended %d bytes", wlen);
 
         return NULL;
 }
 
 static void *
 rxBLOB(sched_task_t *task)
 {
         rpc_cli_t *c = TASK_ARG(task);
         rpc_srv_t *s = c->cli_parent;
         rpc_blob_t *b;          rpc_blob_t *b;
        int ret = 0;        u_char *buf = TASK_DATA(task);
        fd_set fds;        struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
        u_char buf[sizeof(struct tagBLOBHdr)];        int rlen;
        struct tagBLOBHdr *blob;        u_short crc;
         struct timespec ts;
   
        if (!arg) {        FTRACE();
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n"); 
                return NULL; 
        } else 
                s = c->cli_parent; 
   
        do {        /* check for disable service at this moment? */
                /* check for disable service at this moment? */        if (s->srv_blob.state == disable) {
                if (s->srv_blob.state == disable && s->srv_kill != kill) {                usleep(100000);
                        usleep(100000); 
 #ifdef HAVE_PTHREAD_YIELD  #ifdef HAVE_PTHREAD_YIELD
                        pthread_yield();                pthread_yield();
 #endif  #endif
                        continue;                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                }                                TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;
         }
   
                FD_ZERO(&fds);        memset(buf, 0, TASK_DATLEN(task));
                FD_SET(c->cli_sock, &fds);        rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
                ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);        if (rlen == -1) {
                if (ret == -1) {                LOGERR;
                        if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill)                s->srv_blob.state = kill;
                                continue;                return NULL;
         } else if (!rlen || s->srv_kill == kill) { /* receive EOF */
                 s->srv_blob.state = kill;
                 return NULL;
         } else
                 LOGGER("Readed %d bytes", rlen);
   
                        LOGERR;        if (rlen < sizeof(struct tagBLOBHdr)) {
                        ret = -2;                rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
                 schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;
         }
 
         /* check integrity of packet */
         crc = ntohs(blob->hdr_crc);
         blob->hdr_crc ^= blob->hdr_crc;
         if (crc != crcFletcher16((u_short*) buf, ((rlen + 1) & ~1) / 2)) {
                 rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");
                 schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;
         }
 
         /* check RPC packet session info */
         if (rpc_chkPktSession(&blob->hdr_session, &s->srv_session)) {
                 rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                 blob->hdr_cmd = error;
                 goto end;
         } else {
                 /* change socket timeout from last packet */
                 ts.tv_sec = blob->hdr_session.sess_timeout;
                 ts.tv_nsec = 0;
                 schedPolling(TASK_ROOT(task), &ts, NULL);
         }
 
         /* Go to proceed packet ... */
         switch (blob->hdr_cmd) {
                 case get:
                         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);
                                 break;
                         } else
                                 blob->hdr_len = htonl(b->blob_len);
 
                         if (rpc_srv_blobMap(s, b) != -1) {
                                 /* deliver BLOB variable to client */
                                 blob->hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
                                 rpc_srv_blobUnmap(b);
                         } else {
                                 blob->hdr_cmd = error;
                                 blob->hdr_ret = RPC_ERROR(-1);
                         }
                         break;                          break;
                }                case set:
                         if ((b = rpc_srv_registerBLOB(s, ntohl(blob->hdr_len)))) {
                                 /* set new BLOB variable for reply :) */
                                 blob->hdr_var = htonl(b->blob_var);
   
                memset(buf, 0, sizeof buf);                                /* receive BLOB from client */
                ret = recv(c->cli_sock, buf, sizeof buf, 0);                                blob->hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
                if (ret == -1) {                                rpc_srv_blobUnmap(b);
                        LOGERR;                        } else {
                        ret = -3;                                blob->hdr_cmd = error;
                                 blob->hdr_ret = RPC_ERROR(-1);
                         }
                         break;                          break;
                }                case unset:
                /* receive EOF, disable or kill service */                        if (rpc_srv_unregisterBLOB(s, blob->hdr_var) == -1) {
                if (!ret || s->srv_blob.state == kill || s->srv_kill == kill) {                                blob->hdr_cmd = error;
                        ret = 0;                                blob->hdr_ret = RPC_ERROR(-1);
                         }
                         break;                          break;
                }                default:
                if (ret < sizeof(struct tagBLOBHdr)) {                        rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob->hdr_cmd);
                        rpc_SetErr(ERPCMISMATCH, "Error:: too short BLOB packet ...\n");                        blob->hdr_cmd = error;
                        ret = -4;                        blob->hdr_ret = RPC_ERROR(-1);
                        if (s->srv_kill != kill && s->srv_blob.state != kill)        }
                                continue; 
                        else 
                                break; 
                } else 
                        blob = (struct tagBLOBHdr*) buf; 
                /* check BLOB packet session info */ 
                if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session)) { 
                        rpc_SetErr(EINVAL, "Error:: get invalid BLOB session ...\n"); 
                        ret = -5; 
                        goto makeReply; 
                } 
                /* Go to proceed packet ... */ 
                switch (blob->hdr_cmd) { 
                        case get: 
                                if (!(b = rpc_srv_getBLOB(s, blob->hdr_var))) { 
                                        rpc_SetErr(EINVAL, "Error:: var (%x) not found into BLOB server ...\n",  
                                                        blob->hdr_var); 
                                        ret = -6; 
                                        break; 
                                } else 
                                        blob->hdr_len = b->blob_len; 
   
                                if (rpc_srv_blobMap(s, b) != -1) {end:
                                        ret = rpc_srv_sendBLOB(c, b);        schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), 
                                        rpc_srv_blobUnmap(b);                        TASK_DATA(task), TASK_DATLEN(task));
                                } else        schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                                        ret = -7;                        TASK_DATA(task), TASK_DATLEN(task));
                                break;        return NULL;
                        case set:}
                                if ((b = rpc_srv_registerBLOB(s, blob->hdr_len))) { 
                                        /* set new BLOB variable for reply :) */ 
                                        blob->hdr_var = b->blob_var; 
   
                                        ret = rpc_srv_recvBLOB(c, b);static void *
                                        rpc_srv_blobUnmap(b);rpc_srv_dispatchVars(void *arg)
                                } else{
                                        ret = -7;        rpc_cli_t *c = arg;
                                break;        rpc_srv_t *s;
                        case unset:        sched_root_task_t *root;
                                ret = rpc_srv_unregisterBLOB(s, blob->hdr_var);        u_char *buf;
                                if (ret == -1)        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                                        ret = -7; 
                                break; 
                        default: 
                                rpc_SetErr(EPROCUNAVAIL, "Error:: unsupported BLOB command (%d)...\n",  
                                                blob->hdr_cmd); 
                                ret = -7; 
                } 
   
           FTRACE();
   
           if (!arg) {
                   rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
                   return NULL;
           } else
                   s = c->cli_parent;
   
           /* allocate net buffer */
           buf = malloc(sizeof(struct tagBLOBHdr));
           if (!buf) {
                   LOGERR;
                   return NULL;
           }
   
           root = schedBegin();
           if (!root) {
                   rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                   free(buf);
                   return NULL;
           } else {
                   schedTermCondition(root, kill);
                   schedPolling(root, &ts, NULL);
           }
   
           schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
   
           schedRun(root, (void*) &s->srv_blob.state);
           schedEnd(&root);
   
           shutdown(c->cli_sock, SHUT_RDWR);
           close(c->cli_sock);
           memset(c, 0, sizeof(rpc_cli_t));
           free(buf);
           return NULL;
   
   #if 0
 makeReply:  makeReply:
                 /* Replay to client! */                  /* Replay to client! */
                 blob->hdr_cmd = ret < 0 ? error : ok;  
                 blob->hdr_ret = ret;  
                 ret = send(c->cli_sock, buf, sizeof buf, 0);                  ret = send(c->cli_sock, buf, sizeof buf, 0);
                 if (ret == -1) {                  if (ret == -1) {
                         LOGERR;                          LOGERR;
Line 383  makeReply: Line 480  makeReply:
                         else                          else
                                 break;                                  break;
                 }                  }
        } while (ret > -1 || s->srv_kill != kill);#endif
 
        shutdown(c->cli_sock, SHUT_RDWR); 
        close(c->cli_sock); 
        memset(c, 0, sizeof(rpc_cli_t)); 
        return (void*) ((long)ret); 
 }  }
   
 // -------------------------------------------------  // -------------------------------------------------
Line 407  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 499  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
         int n = 1;          int n = 1;
         io_sockaddr_t sa;          io_sockaddr_t sa;
   
           FTRACE();
   
         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 511  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 605  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
         register int i;          register int i;
         rpc_blob_t *f;          rpc_blob_t *f;
   
           FTRACE();
   
         if (!srv) {          if (!srv) {
                 rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");                  rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");
                 return;                  return;
Line 565  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 661  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
         struct timeval tv = { DEF_RPC_TIMEOUT, 0 };          struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
         pthread_attr_t attr;          pthread_attr_t attr;
   
           FTRACE();
   
         if (!srv || srv->srv_blob.state == kill) {          if (!srv || srv->srv_blob.state == kill) {
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");                rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
                 return -1;                  return -1;
         }          }
   
Line 647  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 745  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;
   
           FTRACE();
   
         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 773  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 873  rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
         register int i;          register int i;
         rpc_func_t *f;          rpc_func_t *f;
   
           FTRACE();
   
         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 829  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 931  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
         struct timeval tv = { DEF_RPC_TIMEOUT, 0 };          struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
         pthread_attr_t attr;          pthread_attr_t attr;
   
           FTRACE();
   
         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 907  rpc_srv_execCall(rpc_func_t * __restrict call, struct  Line 1011  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;
   
           FTRACE();
   
         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.6.2.6  
changed lines
  Added in v.1.6.2.8

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