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

version 1.6.2.2, 2012/03/12 15:04:38 version 1.6.2.8, 2012/03/14 15:08:03
Line 46  SUCH DAMAGE. Line 46  SUCH DAMAGE.
 #include "global.h"  #include "global.h"
   
   
   static void *rxPacket(sched_task_t*);
   static void *rxBLOB(sched_task_t*);
   
 static void *  static void *
   txPacket(sched_task_t *task)
   {
           rpc_cli_t *c = TASK_ARG(task);
           rpc_srv_t *s = c->cli_parent;
           rpc_func_t *f = NULL;
           u_char *buf = TASK_DATA(task);
           struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
           int ret, wlen = sizeof(struct tagRPCCall);
           array_t *arr = NULL;
   
           FTRACE();
   
           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;
                   }
           }
   
           /* calculate CRC */
           rpc->call_crc ^= rpc->call_crc;
           rpc->call_crc = htons(crcFletcher16((u_short*) buf, ((wlen + 1) & ~1) / 2));
   
           /* 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);
           else
                   LOGGER("Sended %d bytes", ret);
   
           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);
   
           FTRACE();
   
           /* 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 {
                   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));
                   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 timespec ts;
   
           FTRACE();
   
           memset(buf, 0, TASK_DATLEN(task));
           rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
           if (rlen == -1) {
                   LOGERR;
                   s->srv_kill = s->srv_blob.state = kill;
                   return NULL;
           } else if (!rlen) {     /* receive EOF */
                   s->srv_kill = s->srv_blob.state = kill;
                   return NULL;
           } else
                   LOGGER("Readed %d bytes", rlen);
   
           if (rlen < sizeof(struct tagRPCCall)) {
                   rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
   
                   schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                                   TASK_DATA(task), TASK_DATLEN(task));
                   return NULL;
           } else
                   rpc = (struct tagRPCCall*) buf;
   
           /* check integrity of packet */
           crc = ntohs(rpc->call_crc);
           rpc->call_crc ^= rpc->call_crc;
           if (crc != crcFletcher16((u_short*) buf, ((rlen + 1) & ~1) / 2)) {
                   rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
   
                   schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task), 
                                   TASK_DATA(task), TASK_DATLEN(task));
                   return NULL;
           }
   
           /* check RPC packet session info */
           if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
                   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);
                   goto end;
           } else {
                   /* change socket timeout from last packet */
                   ts.tv_sec = rpc->call_session.sess_timeout;
                   ts.tv_nsec = 0;
                   schedPolling(TASK_ROOT(task), &ts, NULL);
           }
   
           /* execute RPC call */
           schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), 0, 
                           TASK_DATA(task), TASK_DATLEN(task));
   
   end:
           /* 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));
           /* 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 *
 rpc_srv_dispatchCall(void *arg)  rpc_srv_dispatchCall(void *arg)
 {  {
         rpc_cli_t *c = arg;          rpc_cli_t *c = arg;
         rpc_srv_t *s;          rpc_srv_t *s;
         rpc_func_t *f = NULL;  
         array_t *arr;  
         struct tagRPCCall *rpc;  
         struct tagRPCRet *rrpc;  
         rpc_sess_t ses = { 0 };  
         fd_set fds;  
         u_char *buf;          u_char *buf;
        int ret, argc = 0, Limit = 0;        sched_root_task_t *root;
        register int i;        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
        uint16_t tag = 0; 
        uint32_t hash = 0; 
   
           FTRACE();
   
         if (!arg) {          if (!arg) {
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n");                rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
                 return NULL;                  return NULL;
         } else          } else
                 s = c->cli_parent;                  s = c->cli_parent;
   
           /* allocate net buffer */
         buf = malloc(s->srv_netbuf);          buf = malloc(s->srv_netbuf);
         if (!buf) {          if (!buf) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         }          }
   
        do {        root = schedBegin();
                FD_ZERO(&fds);        if (!root) {
                FD_SET(c->cli_sock, &fds);                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);                free(buf);
                if (ret == -1) {                return NULL;
                        if (errno == EINTR && s->srv_kill != kill)        } else {
                                continue;                schedTermCondition(root, kill);
                 schedPolling(root, &ts, NULL);
         }
   
                        LOGERR;        schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);
                        ret = -2; 
                        break; 
                } 
                memset(buf, 0, s->srv_netbuf); 
                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;        schedRun(root, (void*) &s->srv_kill);
                hash = rpc->call_hash;        schedEnd(&root);
   
                /* RPC is OK! Go decapsulate variables ... */        shutdown(c->cli_sock, SHUT_RDWR);
                if (ntohs(rpc->call_argc)) {        close(c->cli_sock);
                        arr = io_buffer2vars(buf + Limit, s->srv_netbuf - Limit,         memset(c, 0, sizeof(rpc_cli_t));
                                        ntohs(rpc->call_argc), 1);        free(buf);
                        if (!arr) {        return NULL;
                                ret = -5;}
                                goto makeReply; 
                        } 
                } else 
                        arr = NULL; 
   
                 /* execute call */  
                 argc = 0;  
                 memcpy(&ses, &rpc->call_session, sizeof ses);  
                 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)static void *
                        io_arrayDestroy(&arr);txBLOB(sched_task_t *task)
 {
         rpc_cli_t *c = TASK_ARG(task);
         u_char *buf = TASK_DATA(task);
         struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
         int wlen = sizeof(struct tagBLOBHdr);
   
makeReply:        FTRACE();
                /* Made reply */ 
                memset(buf, 0, s->srv_netbuf); 
                rrpc = (struct tagRPCRet*) buf; 
                Limit = sizeof(struct tagRPCRet); 
   
                memcpy(&rrpc->ret_session, &ses, sizeof(rpc_sess_t));        /* calculate CRC */
                rrpc->ret_tag = tag;        blob->hdr_crc ^= blob->hdr_crc;
                rrpc->ret_hash = hash;        blob->hdr_crc = htons(crcFletcher16((u_short*) buf, ((wlen + 1) & ~1) / 2));
                rrpc->ret_errno = htonl(rpc_Errno); 
                rrpc->ret_retcode = htonl(ret); 
                rrpc->ret_argc = htons(argc); 
   
                if (argc && arr) {        /* send reply */
                        /* Go Encapsulate variables ... */        wlen = send(TASK_FD(task), buf, wlen, 0);
                        if ((i = io_vars2buffer(buf + Limit, s->srv_netbuf - Limit, arr)) == -1) {        if (wlen == -1)
                                io_clrVars(f->func_vars);                LOGERR;
                                argc = 0;        else if (wlen != sizeof(struct tagBLOBHdr))
                                ret = -7;                rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                                rpc_SetErr(EBADRPC, "Error:: in prepare RPC packet values (-7) ...\n");                                "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
                                goto makeReply;        else
                        } else {                LOGGER("Sended %d bytes", wlen);
                                Limit += i; 
   
                                io_clrVars(f->func_vars);        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;
         u_char *buf = TASK_DATA(task);
         struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
         int rlen;
         u_short crc;
         struct timespec ts;
 
         FTRACE();
 
         /* check for disable service at this moment? */
         if (s->srv_blob.state == disable) {
                 usleep(100000);
 #ifdef HAVE_PTHREAD_YIELD
                 pthread_yield();
 #endif
                 schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                                 TASK_DATA(task), TASK_DATLEN(task));
                 return NULL;
         }
 
         memset(buf, 0, TASK_DATLEN(task));
         rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
         if (rlen == -1) {
                 LOGERR;
                 s->srv_blob.state = kill;
                 return NULL;
         } else if (!rlen || s->srv_kill == kill) { /* receive EOF */
                 s->srv_blob.state = kill;
                 return NULL;
         } else
                 LOGGER("Readed %d bytes", rlen);
 
         if (rlen < sizeof(struct tagBLOBHdr)) {
                 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;
                 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);
   
                ret = send(c->cli_sock, buf, Limit, 0);                                /* receive BLOB from client */
                if (ret == -1) {                                blob->hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
                        LOGERR;                                rpc_srv_blobUnmap(b);
                        ret = -8;                        } else {
                                 blob->hdr_cmd = error;
                                 blob->hdr_ret = RPC_ERROR(-1);
                         }
                         break;                          break;
                }                case unset:
                if (ret != Limit) {                        if (rpc_srv_unregisterBLOB(s, blob->hdr_var) == -1) {
                        rpc_SetErr(EPROCUNAVAIL, "Error:: in send RPC request, should be send %d bytes, "                                blob->hdr_cmd = error;
                                        "really is %d\n", Limit, ret);                                blob->hdr_ret = RPC_ERROR(-1);
                        ret = -9;                        }
                        if (s->srv_kill != kill)                        break;
                                continue;                default:
                        else                        rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob->hdr_cmd);
                                break;                        blob->hdr_cmd = error;
                }                        blob->hdr_ret = RPC_ERROR(-1);
        } while (ret > -1 || s->srv_kill != kill);        }
   
        shutdown(c->cli_sock, SHUT_RDWR);end:
        close(c->cli_sock);        schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), 
        memset(c, 0, sizeof(rpc_cli_t));                        TASK_DATA(task), TASK_DATLEN(task));
        free(buf);        schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
        return (void*) (long)ret;                        TASK_DATA(task), TASK_DATLEN(task));
         return NULL;
 }  }
   
   
 static void *  static void *
 rpc_srv_dispatchVars(void *arg)  rpc_srv_dispatchVars(void *arg)
 {  {
         rpc_cli_t *c = arg;          rpc_cli_t *c = arg;
         rpc_srv_t *s;          rpc_srv_t *s;
        rpc_blob_t *b;        sched_root_task_t *root;
        int ret = 0;        u_char *buf;
        fd_set fds;        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
        u_char buf[sizeof(struct tagBLOBHdr)]; 
        struct tagBLOBHdr *blob; 
   
           FTRACE();
   
         if (!arg) {          if (!arg) {
                rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n");                rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
                 return NULL;                  return NULL;
         } else          } else
                 s = c->cli_parent;                  s = c->cli_parent;
   
        do {        /* allocate net buffer */
                /* check for disable service at this moment? */        buf = malloc(sizeof(struct tagBLOBHdr));
                if (s->srv_blob.state == disable && s->srv_kill != kill) {        if (!buf) {
                        usleep(100000);                LOGERR;
#ifdef HAVE_PTHREAD_YIELD                return NULL;
                        pthread_yield();        }
#endif 
                        continue; 
                } 
   
                FD_ZERO(&fds);        root = schedBegin();
                FD_SET(c->cli_sock, &fds);        if (!root) {
                ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                if (ret == -1) {                free(buf);
                        if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill)                return NULL;
                                continue;        } else {
                 schedTermCondition(root, kill);
                 schedPolling(root, &ts, NULL);
         }
   
                        LOGERR;        schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
                        ret = -2; 
                        break; 
                } 
   
                memset(buf, 0, sizeof buf);        schedRun(root, (void*) &s->srv_blob.state);
                ret = recv(c->cli_sock, buf, sizeof buf, 0);        schedEnd(&root);
                if (ret == -1) { 
                        LOGERR; 
                        ret = -3; 
                        break; 
                } 
                /* receive EOF, disable or kill service */ 
                if (!ret || s->srv_blob.state == kill || s->srv_kill == kill) { 
                        ret = 0; 
                        break; 
                } 
                if (ret < sizeof(struct tagBLOBHdr)) { 
                        rpc_SetErr(ERPCMISMATCH, "Error:: too short BLOB packet ...\n"); 
                        ret = -4; 
                        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) {        shutdown(c->cli_sock, SHUT_RDWR);
                                        ret = rpc_srv_sendBLOB(c, b);        close(c->cli_sock);
                                        rpc_srv_blobUnmap(b);        memset(c, 0, sizeof(rpc_cli_t));
                                } else        free(buf);
                                        ret = -7;        return NULL;
                                break; 
                        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);#if 0
                                        rpc_srv_blobUnmap(b); 
                                } else 
                                        ret = -7; 
                                break; 
                        case unset: 
                                ret = rpc_srv_unregisterBLOB(s, blob->hdr_var); 
                                if (ret == -1) 
                                        ret = -7; 
                                break; 
                        default: 
                                rpc_SetErr(EPROCUNAVAIL, "Error:: unsupported BLOB command (%d)...\n",  
                                                blob->hdr_cmd); 
                                ret = -7; 
                } 
 
 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 324  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); 
 }  }
   
 // -------------------------------------------------  // -------------------------------------------------
   
 /*  /*
 * 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 347  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 361  rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s Line 515  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 532  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 541  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 576  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 593  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 450  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 461  rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) Line 618  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 645  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 504  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 566  rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) Line 725  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 585  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 645  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 807  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 824  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 837  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 846  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 861  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 723  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 730  rpc_srv_endServer(rpc_srv_t ** __restrict psrv) Line 882  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 901  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 915  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 779  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;
         }          }
   
           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 795  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 951  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
         pthread_attr_init(&attr);          pthread_attr_init(&attr);
         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);          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 826  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 983  rpc_srv_loopServer(rpc_srv_t * __restrict srv)
                 } else                  } else
                         c->cli_parent = srv;                          c->cli_parent = srv;
   
                   /* spawn rpc client dispatcher */
                 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 840  rpc_srv_loopServer(rpc_srv_t * __restrict srv) Line 998  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 853  rpc_srv_execCall(rpc_func_t * __restrict call, struct  Line 1012  rpc_srv_execCall(rpc_func_t * __restrict call, struct 
         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, "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.2  
changed lines
  Added in v.1.6.2.8

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