[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.1.1.1.2.1 and 1.1.1.1.2.4

version 1.1.1.1.2.1, 2010/06/18 13:36:01 version 1.1.1.1.2.4, 2010/06/23 11:33:38
Line 12 Line 12
 static void *  static void *
 rpc_srv_dispatchCall(void *arg)  rpc_srv_dispatchCall(void *arg)
 {  {
        rpc_cli_t *c = arg;        rpc_cli_t cli, *c = arg;
         rpc_srv_t *s;          rpc_srv_t *s;
         rpc_val_t *vals, *v = NULL;          rpc_val_t *vals, *v = NULL;
         rpc_func_t *f;          rpc_func_t *f;
Line 74  rpc_srv_dispatchCall(void *arg) Line 74  rpc_srv_dispatchCall(void *arg)
                                                 v[i].val.string = (int8_t*) data;                                                  v[i].val.string = (int8_t*) data;
                                                 data += v[i].val_len + 1;                                                  data += v[i].val_len + 1;
                                                 break;                                                  break;
                                        case array:                                        case blob:
                                                v[i].val.array = (int8_t**) data;                                                rpc_srv_getBLOB(&v[i], data);
                                                data += v[i].val_len;                                                data += sizeof(rpc_cli_t);
                                                 break;                                                  break;
                                         default:                                          default:
                                                 break;                                                  break;
Line 135  rpc_srv_dispatchCall(void *arg) Line 135  rpc_srv_dispatchCall(void *arg)
                                                 data += vals[i].val_len + 1;                                                  data += vals[i].val_len + 1;
                                                 Limit += vals[i].val_len + 1;                                                  Limit += vals[i].val_len + 1;
                                                 break;                                                  break;
                                        case array:                                        case blob:
                                                if (ret || Limit + vals[i].val_len > BUFSIZ) {                                                if (ret || Limit + sizeof(rpc_cli_t) > BUFSIZ) {
                                                         rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");                                                          rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
                                                         rrpc.ret_retcode = ret = -7;                                                          rrpc.ret_retcode = ret = -7;
                                                         rrpc.ret_argc = 0;                                                          rrpc.ret_argc = 0;
                                                         break;                                                          break;
                                                 }                                                  }
   
                                                memcpy(data, vals[i].val.array, vals[i].val_len);                                                memcpy(data, &cli, sizeof(rpc_cli_t));
                                                data += vals[i].val_len;                                                data += sizeof(rpc_cli_t);
                                                Limit += vals[i].val_len;                                                Limit += sizeof(rpc_cli_t);
 
                                                 rpc_srv_setBLOB(&vals[i], &cli);
                                                 break;                                                  break;
                                         default:                                          default:
                                                 break;                                                  break;
Line 174  rpc_srv_dispatchCall(void *arg) Line 176  rpc_srv_dispatchCall(void *arg)
         return (void*) ret;          return (void*) ret;
 }  }
   
   
   static void *
   rpc_srv_dispatchVars(void *arg)
   {
           rpc_cli_t *c = arg;
   //      rpc_srv_t *s;
           int ret;
   
           return (void*) ret;
   }
   
 // -------------------------------------------------  // -------------------------------------------------
   
 /*  /*
    * rpc_srv_initBLOBServer() Init & create BLOB Server
    * @Port = Port for bind server, if Port == 0 default port is selected
    * return: -1 == error or 0 bind and created BLOB server instance
    */
   int
   rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port)
   {
           int n = 1;
           struct sockaddr_in sin;
           struct sockaddr_in6 sin6;
   
           if (!srv) {
                   rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init BLOB server ...\n");
                   return -1;
           }
           if (srv->srv_blob.state) {
                   rpc_SetErr(EPERM, "Warning:: Already started BLOB server!\n");
                   return 0;
           }
           if (!Port)
                   Port = RPC_DEFPORT + 1;
   
           memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
           srv->srv_blob.server.cli_tid = pthread_self();
           srv->srv_blob.server.cli_parent = srv;
           if (srv->srv_server.cli_sa.sa_family == AF_INET) {
                   memcpy(&sin, &srv->srv_server.cli_sa, sizeof sin);
                   sin.sin_port = htons(Port);
                   memcpy(&srv->srv_blob.server.cli_sa, &sin, sizeof(struct sockaddr));
           } else {
                   memcpy(&sin6, &srv->srv_server.cli_sa, sizeof sin6);
                   sin6.sin6_port = htons(Port);
                   memcpy(&srv->srv_blob.server.cli_sa, &sin6, sizeof(struct sockaddr));
           }
   
           srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0);
           if (srv->srv_blob.server.cli_sock == -1) {
                   LOGERR;
                   return -1;
           }
           if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                   LOGERR;
                   close(srv->srv_blob.server.cli_sock);
                   return -1;
           }
           if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa, 
                                   sizeof srv->srv_blob.server.cli_sa) == -1) {
                   LOGERR;
                   close(srv->srv_blob.server.cli_sock);
                   return -1;
           }
   
           srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
           if (!srv->srv_blob.clients) {
                   LOGERR;
                   close(srv->srv_blob.server.cli_sock);
                   return -1;
           } else
                   memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
   
           pthread_mutex_init(&srv->srv_blob.mtx, NULL);
   
           pthread_mutex_lock(&srv->srv_mtx);
           rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
           rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0);
           rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0);
           pthread_mutex_unlock(&srv->srv_mtx);
   
           srv->srv_blob.state = 1;        // enable BLOB
           return 0;
   }
   
   /*
    * rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources
    * @srv = RPC Server instance
    * return: none
    */
   void
   rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
   {
           rpc_cli_t *c;
           register int i;
           rpc_blob_t *f;
   
           if (!srv) {
                   rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
                   return;
           } else
                   srv->srv_blob.state = 0;
   
           for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
                   if (c->cli_sa.sa_family)
                           shutdown(c->cli_sock, SHUT_RDWR);
           close(srv->srv_blob.server.cli_sock);
   
           if (srv->srv_blob.clients)
                   free(srv->srv_blob.clients);
   
           pthread_mutex_lock(&srv->srv_blob.mtx);
           while ((f = srv->srv_blob.blobs)) {
                   srv->srv_blob.blobs = f->blob_next;
                   free(f);
           }
           pthread_mutex_unlock(&srv->srv_blob.mtx);
   
           pthread_mutex_destroy(&srv->srv_blob.mtx);
   }
   
   /*
    * rpc_srv_execBLOBServer() Execute Main BLOB server loop and wait for clients requests
    * @srv = RPC Server instance
    * return: -1 error or 0 ok, infinite loop ...
    */
   int
   rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)
   {
           socklen_t salen = sizeof(struct sockaddr);
           register int i;
           rpc_cli_t *c;
           fd_set fds;
           int ret;
           struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
   
           if (!srv || !srv->srv_blob.state) {
                   rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");
                   return -1;
           }
   
           if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
                   LOGERR;
                   return -1;
           }
   
           while (!rpc_Kill) {
                   for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
                           if (!c->cli_sa.sa_family)
                                   break;
                   if (c && c->cli_sa.sa_family && c->cli_parent) {
                           usleep(1000000);
                           continue;
                   }
   
                   FD_ZERO(&fds);
                   FD_SET(srv->srv_blob.server.cli_sock, &fds);
                   ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
                   if (ret == -1) {
                           LOGERR;
                           ret = 1;
                           break;
                   }
                   if (!ret)
                           continue;
   
                   c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen);
                   if (c->cli_sock == -1) {
                           LOGERR;
                           continue;
                   } else
                           c->cli_parent = srv;
   
                   if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) {
                           LOGERR;
                           continue;
                   }
           }
   
           return 0;
   }
   
   
   /*
  * 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
Line 243  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 427  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         srv->srv_session.sess_program = regProgID;          srv->srv_session.sess_program = regProgID;
         srv->srv_session.sess_process = regProcID;          srv->srv_session.sess_process = regProcID;
   
           srv->srv_server.cli_tid = pthread_self();
         srv->srv_server.cli_parent = srv;          srv->srv_server.cli_parent = srv;
         if (family == AF_INET)          if (family == AF_INET)
                 memcpy(&srv->srv_server.cli_sa, &sin, sizeof srv->srv_server.cli_sa);                  memcpy(&srv->srv_server.cli_sa, &sin, sizeof srv->srv_server.cli_sa);
Line 281  rpc_srv_initServer(u_int regProgID, u_int regProcID, i Line 466  rpc_srv_initServer(u_int regProgID, u_int regProcID, i
         rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);          rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);
         rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);          rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);
   
        pthread_mutex_init(&rpc_mtx, NULL);        pthread_mutex_init(&srv->srv_mtx, NULL);
         return srv;          return srv;
 }  }
   
Line 302  rpc_srv_endServer(rpc_srv_t * __restrict srv) Line 487  rpc_srv_endServer(rpc_srv_t * __restrict srv)
                 return;                  return;
         }          }
   
        pthread_mutex_destroy(&rpc_mtx);        rpc_srv_endBLOBServer(srv);
   
         while ((f = srv->srv_funcs)) {  
                 srv->srv_funcs = f->func_next;  
                 free(f);  
         }  
   
         for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++)          for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++)
                 if (c->cli_sa.sa_family)                  if (c->cli_sa.sa_family)
                         shutdown(c->cli_sock, SHUT_RDWR);                          shutdown(c->cli_sock, SHUT_RDWR);
Line 319  rpc_srv_endServer(rpc_srv_t * __restrict srv) Line 499  rpc_srv_endServer(rpc_srv_t * __restrict srv)
                 srv->srv_numcli = 0;                  srv->srv_numcli = 0;
         }          }
   
           pthread_mutex_lock(&srv->srv_mtx);
           while ((f = srv->srv_funcs)) {
                   srv->srv_funcs = f->func_next;
                   free(f);
           }
           pthread_mutex_unlock(&srv->srv_mtx);
   
           pthread_mutex_destroy(&srv->srv_mtx);
   
         free(srv);          free(srv);
         srv = NULL;          srv = NULL;
 }  }
Line 334  rpc_srv_execServer(rpc_srv_t * __restrict srv) Line 523  rpc_srv_execServer(rpc_srv_t * __restrict srv)
         socklen_t salen = sizeof(struct sockaddr);          socklen_t salen = sizeof(struct sockaddr);
         register int i;          register int i;
         rpc_cli_t *c;          rpc_cli_t *c;
           fd_set fds;
           int ret;
           struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
   
         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");
Line 353  rpc_srv_execServer(rpc_srv_t * __restrict srv) Line 545  rpc_srv_execServer(rpc_srv_t * __restrict srv)
                         usleep(1000000);                          usleep(1000000);
                         continue;                          continue;
                 }                  }
   
                   FD_ZERO(&fds);
                   FD_SET(srv->srv_server.cli_sock, &fds);
                   ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
                   if (ret == -1) {
                           LOGERR;
                           ret = 1;
                           break;
                   }
                   if (!ret)
                           continue;
   
                 c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen);                  c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen);
                 if (c->cli_sock == -1) {                  if (c->cli_sock == -1) {
                         printf("%s(%d): #%d - %s\n", __func__, __LINE__, errno, strerror(errno));  
                         LOGERR;                          LOGERR;
                         continue;                          continue;
                 } else                  } else
Line 367  rpc_srv_execServer(rpc_srv_t * __restrict srv) Line 570  rpc_srv_execServer(rpc_srv_t * __restrict srv)
                 }                  }
         }          }
   
         /* not reached !!! */  
         return 0;          return 0;
 }  }
   
Line 521  rpc_srv_registerCall(rpc_srv_t * __restrict srv, const Line 723  rpc_srv_registerCall(rpc_srv_t * __restrict srv, const
                 return -1;                  return -1;
         }          }
   
           pthread_mutex_lock(&srv->srv_mtx);
         func->func_next = srv->srv_funcs;          func->func_next = srv->srv_funcs;
         srv->srv_funcs = func;          srv->srv_funcs = func;
           pthread_mutex_unlock(&srv->srv_mtx);
         return 0;          return 0;
 }  }
   
Line 560  rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con Line 764  rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con
         if (!f)         // not found element for unregister          if (!f)         // not found element for unregister
                 return 0;                  return 0;
   
           pthread_mutex_lock(&srv->srv_mtx);
         if (srv->srv_funcs == f) {      // if is 1st element          if (srv->srv_funcs == f) {      // if is 1st element
                 srv->srv_funcs = srv->srv_funcs->func_next;                  srv->srv_funcs = srv->srv_funcs->func_next;
   
Line 574  rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con Line 779  rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con
                         free(f->func_vals);                          free(f->func_vals);
                 free(f);                  free(f);
         }          }
           pthread_mutex_unlock(&srv->srv_mtx);
   
         return 1;          return 1;
 }  }
Removed from v.1.1.1.1.2.1  
changed lines
  Added in v.1.1.1.1.2.4

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