#include <tcp_srv.hh>
Inheritance diagram for CMS_SERVER_REMOTE_TCP_PORT:
|
|
Definition at line 221 of file tcp_srv.cc. 00222 {
00223 taskDelete (id);
00224 }
00225 #endif
00226 #if defined(NO_THREADS) || defined(SGI)
00227 kill (id, SIGINT);
00228 waitpid (id, NULL, 0);
00229 #endif
00230 #ifdef WIN32
00231 TerminateThread ((HANDLE) id, -1);
00232 #endif
00233 tcpsvr_threads_killed++;
00234 }
00235
00236 void
00237 CMS_SERVER_REMOTE_TCP_PORT::unregister_port ()
00238 {
00239 CLIENT_TCP_PORT *client;
00240 int number_of_connected_clients = 0;
00241 #if defined(sunos5) || defined(VXWORKS)
00242 client = (CLIENT_TCP_PORT *) client_ports->get_head ();
00243 while (NULL != client)
00244 {
00245 if (client->threadId > 0 && client->blocking)
00246 {
00247 blocking_thread_kill (client->threadId);
00248 client->threadId = 0;
00249 }
00250 client = (CLIENT_TCP_PORT *) client_ports->get_next ();
00251 }
00252 #endif
00253
00254 client = (CLIENT_TCP_PORT *) client_ports->get_head ();
00255 while (NULL != client)
|
|
|
Definition at line 257 of file tcp_srv.cc. 00264 {
00265 delete client;
|
|
|
Reimplemented from CMS_SERVER_REMOTE_PORT. Definition at line 364 of file tcp_srv.cc. 00365 {
00366 #ifndef UNDER_CE
00367 rcs_print_error ("bind error: %d -- %s\n", errno, strerror (errno));
00368 #endif
00369 rcs_print_error
00370 ("Server can not bind the connection socket on port %d.\n",
00371 dl_ntohs (server_socket_address.sin_port));
00372 return;
00373 }
00374 if (dl_listen (connection_socket, 5) < 0)
00375 {
00376 #ifndef UNDER_CE
00377 rcs_print_error ("listen error: %d -- %s\n", errno, strerror (errno));
00378 #endif
00379 rcs_print_error ("TCP Server: error on call to listen for port %d.\n",
00380 dl_ntohs (server_socket_address.sin_port));
00381 return;
00382 }
00383 port_registered = 1;
00384
00385 }
00386
00387 static int last_pipe_signum = 0;
00388
00389 static void
00390 handle_pipe_error (int signum)
00391 {
00392 last_pipe_signum = signum;
00393 rcs_print_error ("SIGPIPE intercepted.\n");
00394 }
00395
00396 void
00397 CMS_SERVER_REMOTE_TCP_PORT::run ()
00398 {
00399 unsigned long bytes_ready;
00400 int ready_descriptors;
00401 if (NULL == client_ports)
00402 {
00403 rcs_print_error ("CMS_SERVER: List of client ports is NULL.\n");
00404 return;
00405 }
|
|
|
Reimplemented from CMS_SERVER_REMOTE_PORT. Definition at line 466 of file tcp_srv.cc. 00479 {
00480 rcs_print_debug (PRINT_SOCKET_CONNECT,
00481 "Socket closed by host with IP address %s.\n",
00482 dl_inet_ntoa
00483 (client_port_to_check->address.sin_addr));
00484 if (NULL != client_port_to_check->subscriptions)
00485 {
00486 TCP_CLIENT_SUBSCRIPTION_INFO *clnt_sub_info =
00487 (TCP_CLIENT_SUBSCRIPTION_INFO *)
00488 client_port_to_check->subscriptions->get_head ();
00489 while (NULL != clnt_sub_info)
00490 {
00491 if (NULL != clnt_sub_info->sub_buf_info &&
00492 clnt_sub_info->subscription_list_id >= 0)
00493 {
00494 if (NULL !=
00495 clnt_sub_info->sub_buf_info->sub_clnt_info)
00496 {
00497 clnt_sub_info->sub_buf_info->sub_clnt_info->
00498 delete_node (clnt_sub_info->
00499 subscription_list_id);
00500 if (clnt_sub_info->sub_buf_info->
00501 sub_clnt_info->list_size < 1)
00502 {
00503 delete clnt_sub_info->sub_buf_info->
00504 sub_clnt_info;
00505 clnt_sub_info->sub_buf_info->
00506 sub_clnt_info = NULL;
00507 if (NULL != subscription_buffers
00508 && clnt_sub_info->sub_buf_info->
00509 list_id >= 0)
00510 {
00511 subscription_buffers->delete_node
00512 (clnt_sub_info->sub_buf_info->
00513 list_id);
00514 delete clnt_sub_info->sub_buf_info;
00515 clnt_sub_info->sub_buf_info = NULL;
00516 }
00517 }
00518 clnt_sub_info->sub_buf_info = NULL;
00519 }
00520 delete clnt_sub_info;
00521 clnt_sub_info =
00522 (TCP_CLIENT_SUBSCRIPTION_INFO *)
00523 client_port_to_check->subscriptions->
00524 get_next ();
00525 }
00526 delete client_port_to_check->subscriptions;
00527 client_port_to_check->subscriptions = NULL;
00528 recalculate_polling_interval ();
00529 }
00530 }
00531 if (client_port_to_check->threadId > 0
00532 && client_port_to_check->blocking)
00533 {
00534 blocking_thread_kill (client_port_to_check->threadId);
00535 }
00536 dl_closesocket (client_port_to_check->socket_fd);
00537 RCS_FD_CLR (client_port_to_check->socket_fd, &read_fd_set);
00538 client_port_to_check->socket_fd = -1;
00539 delete client_port_to_check;
00540 client_ports->delete_current_node ();
00541 }
00542 else
00543 {
00544 if (client_port_to_check->blocking)
00545 {
00546 if (client_port_to_check->threadId > 0)
00547 {
00548 rcs_print_debug (PRINT_SERVER_THREAD_ACTIVITY,
00549 "Data recieved from %s:%d when it should be blocking (bytes_ready=%d).\n",
00550 dl_inet_ntoa
00551 (client_port_to_check->address.
00552 sin_addr),
00553 client_port_to_check->socket_fd,
00554 bytes_ready);
00555 rcs_print_debug (PRINT_SERVER_THREAD_ACTIVITY,
00556 "Killing handler %d.\n",
00557 client_port_to_check->threadId);
00558
00559 blocking_thread_kill
00560 (client_port_to_check->threadId);
00561 #if 0
00562 *((u_long *) temp_buffer) =
00563 dl_htonl (client_port_to_check->serial_number);
00564 *((u_long *) temp_buffer + 1) =
00565 dl_htonl ((unsigned long) CMS_SERVER_SIDE_ERROR);
00566 *((u_long *) temp_buffer + 2) = dl_htonl (0); /* size */
00567 *((u_long *) temp_buffer + 3) = dl_htonl (0); /* write_id */
00568 *((u_long *) temp_buffer + 4) = dl_htonl (0); /* was_read */
00569 sendn (client_port_to_check->socket_fd, temp_buffer,
00570 20, 0, dtimeout);
00571 #endif
00572 client_port_to_check->threadId = 0;
00573 client_port_to_check->blocking = 0;
00574 }
00575 }
00576 handle_request (client_port_to_check);
00577 }
00578 ready_descriptors--;
00579 }
00580 else
00581 {
00582 RCS_FD_SET (client_port_to_check->socket_fd, &read_fd_set);
00583 }
00584 client_port_to_check =
00585 (CLIENT_TCP_PORT *) client_ports->get_next ();
00586 }
00587 if (dl_fd_isset (connection_socket, &read_fd_set)
00588 && ready_descriptors > 0)
00589 {
00590 ready_descriptors--;
00591 int client_address_length;
00592 new_client_port = new CLIENT_TCP_PORT ();
00593 client_address_length = sizeof (new_client_port->address);
00594 new_client_port->socket_fd = dl_accept (connection_socket,
00595 (struct sockaddr *)
00596 &new_client_port->address,
00597 &client_address_length);
00598 current_clients++;
00599 if (current_clients > max_clients)
00600 {
00601 max_clients = current_clients;
00602 }
00603 if (new_client_port->socket_fd < 0)
00604 {
00605 #ifndef UNDER_CE
00606 rcs_print_error ("server: accept error -- %d %s \n", errno,
00607 strerror (errno));
00608 #endif
00609 continue;
00610 }
00611 rcs_print_debug (PRINT_SOCKET_CONNECT,
00612 "Socket opened by host with IP address %s.\n",
00613 dl_inet_ntoa (new_client_port->address.sin_addr));
00614 new_client_port->serial_number = 0;
00615 new_client_port->blocking = 0;
00616 if (NULL != client_ports)
00617 {
00618 client_ports->store_at_tail (new_client_port,
00619 sizeof (new_client_port), 0);
00620 }
00621 if (maxfdpl < new_client_port->socket_fd + 1)
00622 {
00623 maxfdpl = new_client_port->socket_fd + 1;
00624 }
00625 RCS_FD_SET (new_client_port->socket_fd, &read_fd_set);
00626 }
00627 else
00628 {
00629 RCS_FD_SET (connection_socket, &read_fd_set);
00630 }
00631 if (0 != ready_descriptors)
00632 {
00633 rcs_print_error ("%d descriptors ready but not serviced.\n",
00634 ready_descriptors);
00635 }
00636 update_subscriptions ();
00637 }
00638 }
00639
00640 static int tcpsvr_handle_blocking_request_sigint_count = 0;
00641 static int tcpsvr_last_sig = 0;
00642
00643 void
00644 tcpsvr_handle_blocking_request_sigint_handler (int sig)
00645 {
00646 tcpsvr_last_sig = sig;
00647 tcpsvr_handle_blocking_request_sigint_count++;
00648 }
00649
00650 #if defined(sunos5) || defined(VXWORKS) || defined(POSIX_THREADS) || defined(NO_THREADS) || defined(WIN32) || defined(SGI)
00651 #ifdef VXWORKS
00652 int
00653 tcpsvr_handle_blocking_request (void *_req)
00654 #else
00655 #ifdef WIN32
00656 #ifndef UNDER_CE
00657 void *__cdecl
00658 tcpsvr_handle_blocking_request (void *_req)
00659 #else
00660 unsigned long __stdcall
00661 tcpsvr_handle_blocking_request (void *_req)
00662 #endif
00663 #else
00664 #ifdef SGI
00665 void
00666 tcpsvr_handle_blocking_request (void *_req)
00667 #else
00668 void *
00669 tcpsvr_handle_blocking_request (void *_req)
00670 #endif
00671 #endif
00672 #endif
00673 {
00674 #ifndef UNDER_CE
00675 signal (SIGINT, tcpsvr_handle_blocking_request_sigint_handler);
00676 #endif
00677 TCPSVR_BLOCKING_READ_REQUEST *blocking_read_req =
00678 (TCPSVR_BLOCKING_READ_REQUEST *) _req;
00679 char temp_buffer[0x2000];
00680 if (_req == NULL)
00681 {
00682 tcpsvr_threads_returned_early++;
00683 #ifndef SGI
00684 return 0;
00685 #else
00686 return;
00687 #endif
00688 }
00689 double dtimeout =
00690 ((double) (blocking_read_req->timeout_millis + 10)) / 1000.0;
00691 if (dtimeout < 0)
00692 {
00693 dtimeout = 600.0;
00694 }
00695 if (dtimeout < 0.5)
00696 {
00697 dtimeout = 0.5;
00698 }
00699 if (dtimeout > 600.0)
00700 {
00701 dtimeout = 600.0;
00702 }
00703 CLIENT_TCP_PORT *_client_tcp_port = blocking_read_req->_client_tcp_port;
00704 CMS_SERVER *server = blocking_read_req->server;
00705
00706 if (NULL == server || NULL == _client_tcp_port)
00707 {
|
|
|
Reimplemented from CMS_SERVER_REMOTE_PORT. Definition at line 408 of file tcp_srv.cc. 00426 {
00427 if (polling_enabled)
00428 {
00429 memcpy (&read_fd_set_copy, &read_fd_set, sizeof (fd_set));
00430 memcpy (&write_fd_set_copy, &write_fd_set, sizeof (fd_set));
00431 ready_descriptors =
00432 dl_select (maxfdpl, &read_fd_set, &write_fd_set, (fd_set *) NULL,
00433 (timeval *) & select_timeout);
00434 if (ready_descriptors == 0)
00435 {
00436 update_subscriptions ();
00437 memcpy (&read_fd_set, &read_fd_set_copy, sizeof (fd_set));
00438 memcpy (&write_fd_set, &write_fd_set_copy, sizeof (fd_set));
00439 continue;
00440 }
00441 }
00442 else
00443 {
00444 ready_descriptors =
00445 dl_select (maxfdpl, &read_fd_set, &write_fd_set, (fd_set *) NULL,
00446 (timeval *) NULL);
00447
00448 }
00449 #ifndef UNDER_CE
00450 if (ready_descriptors < 0)
00451 {
00452 rcs_print_error ("server: select error.(errno = %d | %s)\n",
00453 errno, strerror (errno));
00454 }
|
|
|
Reimplemented from CMS_SERVER_REMOTE_PORT. Definition at line 306 of file tcp_srv.cc. 00306 {
00307 if (min_compatible_version < 1e-6 ||
00308 (min_compatible_version > _cms->min_compatible_version &&
00309 _cms->min_compatible_version > 1e-6))
00310 {
00311 min_compatible_version = _cms->min_compatible_version;
00312 }
00313 if (_cms->confirm_write)
00314 {
00315 confirm_write = _cms->confirm_write;
00316 }
00317 }
00318 if (_cms->total_subdivisions > max_total_subdivisions)
00319 {
00320 max_total_subdivisions = _cms->total_subdivisions;
00321 }
00322 if (server_socket_address.sin_port == 0)
00323 {
00324 server_socket_address.sin_port =
00325 dl_htons (((u_short) _cms->tcp_port_number));
00326 port_num = _cms->tcp_port_number;
00327 return 1;
00328 }
00329 if (server_socket_address.sin_port ==
00330 dl_htons (((u_short) _cms->tcp_port_number)))
00331 {
00332 port_num = _cms->tcp_port_number;
00333 return 1;
00334 }
00335 return 0;
00336 }
00337
00338 void
00339 CMS_SERVER_REMOTE_TCP_PORT::register_port ()
00340 {
00341 port_registered = 0;
00342 rcs_print_debug (PRINT_CMS_CONFIG_INFO,
00343 "Registering server on TCP port %d.\n",
00344 dl_ntohs (server_socket_address.sin_port));
00345 if (server_socket_address.sin_port == 0)
00346 {
00347 rcs_print_error ("server can not register on port number 0.\n");
00348 return;
00349 }
00350 if ((connection_socket = dl_socket (AF_INET, SOCK_STREAM, 0)) < 0)
00351 {
00352 #ifndef UNDER_CE
00353 rcs_print_error ("socket error: %d -- %s\n", errno, strerror (errno));
00354 #endif
00355 rcs_print_error ("Server can not open stream socket.\n");
00356 return;
00357 }
00358
00359 if (set_tcp_socket_options (connection_socket) < 0)
00360 {
00361 return;
|
|
|
Definition at line 910 of file tcp_srv.cc. 00917 : fd = %d, serial_number=%d, request_type=%d, buffer_number=%d\n",
00918 _client_tcp_port->socket_fd,
00919 _client_tcp_port->serial_number, request_type,
00920 buffer_number);
00921
00922 if (NULL != _client_tcp_port->diag_info)
00923 {
00924 _client_tcp_port->diag_info->buffer_number = buffer_number;
00925 server->set_diag_info (_client_tcp_port->diag_info);
00926 }
00927 else if (server->diag_enabled)
00928 {
00929 server->reset_diag_info (buffer_number);
00930 }
00931
00932 switch_function (_client_tcp_port,
00933 server,
00934 request_type, buffer_number, received_serial_number);
00935
00936 if (NULL != _client_tcp_port->diag_info &&
00937 NULL != server->last_local_port_used && server->diag_enabled)
00938 {
00939 if (NULL != server->last_local_port_used->cms)
00940 {
00941 if (NULL !=
00942 server->last_local_port_used->cms->handle_to_global_data)
00943 {
00944 _client_tcp_port->diag_info->bytes_moved =
00945 server->last_local_port_used->cms->handle_to_global_data->
00946 total_bytes_moved;
00947 }
00948 }
00949 }
00950 }
00951
00952 void
00953 CMS_SERVER_REMOTE_TCP_PORT::switch_function (CLIENT_TCP_PORT *
00954 _client_tcp_port,
00955 CMS_SERVER * server,
00956 long request_type,
00957 long buffer_number,
00958 long received_serial_number)
00959 {
00960 int total_subdivisions = 1;
00961 CLIENT_TCP_PORT *client_port_to_check = NULL;
00962 switch (request_type)
00963 {
00964 case REMOTE_CMS_SET_DIAG_INFO_REQUEST_TYPE:
00965 {
00966 if (NULL == _client_tcp_port->diag_info)
00967 {
00968 _client_tcp_port->diag_info = new REMOTE_SET_DIAG_INFO_REQUEST ();
00969 }
00970 if (recvn
00971 (_client_tcp_port->socket_fd, server->set_diag_info_buf, 68, 0,
00972 -1, NULL) < 0)
00973 {
00974 rcs_print_error ("Can not read from client port (%d) from %s\n",
00975 _client_tcp_port->socket_fd,
00976 dl_inet_ntoa (_client_tcp_port->address.
00977 sin_addr));
00978 _client_tcp_port->errors++;
00979 return;
00980 }
00981 _client_tcp_port->diag_info->bytes_moved = 0.0;
00982 _client_tcp_port->diag_info->buffer_number = buffer_number;
00983 memcpy (_client_tcp_port->diag_info->process_name,
00984 server->set_diag_info_buf, 16);
00985 memcpy (_client_tcp_port->diag_info->host_sysinfo,
00986 server->set_diag_info_buf + 16, 32);
00987 _client_tcp_port->diag_info->pid =
00988 dl_htonl (*((u_long *) (server->set_diag_info_buf + 48)));
00989 _client_tcp_port->diag_info->c_num =
00990 dl_htonl (*((u_long *) (server->set_diag_info_buf + 52)));
00991 memcpy (&(_client_tcp_port->diag_info->rcslib_ver),
00992 server->set_diag_info_buf + 56, 8);
00993 _client_tcp_port->diag_info->reverse_flag =
00994 *((int *) ((char *) server->set_diag_info_buf + 64));
00995 if (_client_tcp_port->diag_info->reverse_flag == 0x44332211)
00996 {
00997 _client_tcp_port->diag_info->rcslib_ver =
00998 (double) tcp_svr_reverse_double ((double)
00999 _client_tcp_port->
01000 diag_info->rcslib_ver);
01001 }
01002 }
01003 break;
01004
01005
01006 case REMOTE_CMS_GET_DIAG_INFO_REQUEST_TYPE:
01007 {
01008 REMOTE_GET_DIAG_INFO_REQUEST diagreq;
01009 diagreq.buffer_number = buffer_number;
01010 REMOTE_GET_DIAG_INFO_REPLY *diagreply = NULL;
01011 diagreply =
01012 (REMOTE_GET_DIAG_INFO_REPLY *) server->process_request (&diagreq);
01013 if (NULL == diagreply)
01014 {
01015 *((u_long *) temp_buffer) =
01016 dl_htonl (_client_tcp_port->serial_number);
01017 *((u_long *) temp_buffer + 1) =
01018 dl_htonl ((unsigned long) CMS_SERVER_SIDE_ERROR);
01019 if (sendn
|
|
|
Definition at line 2107 of file tcp_srv.cc. 02113 {
02114 temp_clnt_info->last_id_read = server->read_reply->write_id;
02115 temp_clnt_info->last_sub_sent_time = cur_time;
02116 temp_clnt_info->clnt_port->serial_number++;
02117 *((u_long *) temp_buffer) =
02118 dl_htonl (temp_clnt_info->clnt_port->serial_number);
02119 if (server->read_reply->size < 0x2000 - 20
02120 && server->read_reply->size > 0)
02121 {
02122 memcpy (temp_buffer + 20, server->read_reply->data,
02123 server->read_reply->size);
02124 if (sendn
02125 (temp_clnt_info->clnt_port->socket_fd, temp_buffer,
02126 20 + server->read_reply->size, 0, dtimeout) < 0)
02127 {
02128 temp_clnt_info->clnt_port->errors++;
02129 return;
02130 }
02131 }
02132 else
02133 {
02134 if (sendn (temp_clnt_info->clnt_port->socket_fd,
02135 temp_buffer, 20, 0, dtimeout) < 0)
02136 {
02137 temp_clnt_info->clnt_port->errors++;
02138 return;
02139 }
02140 if (server->read_reply->size > 0)
02141 {
02142 if (sendn (temp_clnt_info->clnt_port->socket_fd,
02143 server->read_reply->data,
02144 server->read_reply->size, 0, dtimeout) < 0)
02145 {
02146 temp_clnt_info->clnt_port->errors++;
02147 return;
02148 }
02149 }
02150 }
02151 }
02152 if (temp_clnt_info->last_id_read < buf_info->min_last_id)
02153 {
02154 buf_info->min_last_id = temp_clnt_info->last_id_read;
02155 }
02156 temp_clnt_info =
02157 (TCP_CLIENT_SUBSCRIPTION_INFO *)
02158 buf_info->sub_clnt_info->get_next ();
02159 }
02160 buf_info =
02161 (TCP_BUFFER_SUBSCRIPTION_INFO *) subscription_buffers->get_next ();
02162 }
02163 }
02164
02165
02166 TCP_BUFFER_SUBSCRIPTION_INFO::TCP_BUFFER_SUBSCRIPTION_INFO ()
02167 {
02168 buffer_number = -1;
02169 min_last_id = 0;
02170 list_id = -1;
02171 sub_clnt_info = NULL;
02172 }
02173
02174 TCP_BUFFER_SUBSCRIPTION_INFO::~TCP_BUFFER_SUBSCRIPTION_INFO ()
02175 {
02176 buffer_number = -1;
02177 min_last_id = 0;
02178 list_id = -1;
02179 if (NULL != sub_clnt_info)
02180 {
02181 delete sub_clnt_info;
02182 sub_clnt_info = NULL;
02183 }
02184 }
02185
02186 TCP_CLIENT_SUBSCRIPTION_INFO::TCP_CLIENT_SUBSCRIPTION_INFO ()
02187 {
02188 subscription_type = CMS_NO_SUBSCRIPTION;
02189 poll_interval_millis = 30000;
02190 last_sub_sent_time = 0.0;
02191 subscription_list_id = -1;
02192 buffer_number = -1;
02193 subscription_paused = 0;
02194 last_id_read = 0;
02195 sub_buf_info = NULL;
02196 clnt_port = NULL;
02197 }
02198
02199 TCP_CLIENT_SUBSCRIPTION_INFO::~TCP_CLIENT_SUBSCRIPTION_INFO ()
02200 {
02201 subscription_type = CMS_NO_SUBSCRIPTION;
02202 poll_interval_millis = 30000;
02203 last_sub_sent_time = 0.0;
02204 subscription_list_id = -1;
02205 buffer_number = -1;
02206 subscription_paused = 0;
02207 last_id_read = 0;
02208 sub_buf_info = NULL;
02209 clnt_port = NULL;
02210 }
02211
02212
02213 CLIENT_TCP_PORT::CLIENT_TCP_PORT ()
02214 {
02215 serial_number = 0;
02216 errors = 0;
02217 max_errors = 50;
02218 address.sin_port = 0;
02219 address.sin_family = AF_INET;
02220 address.sin_addr.s_addr = dl_htonl (INADDR_ANY);
02221 socket_fd = -1;
02222 subscriptions = NULL;
02223 tid = -1;
02224 pid = -1;
02225 blocking_read_req = NULL;
02226 threadId = -1;
02227 diag_info = NULL;
02228 }
02229
02230 CLIENT_TCP_PORT::~CLIENT_TCP_PORT ()
02231 {
02232 if (socket_fd > 0)
|
|
||||||||||||||||||||
|
Definition at line 1951 of file tcp_srv.cc. 01957 {
01958 TCP_CLIENT_SUBSCRIPTION_INFO *temp_clnt_info =
01959 (TCP_CLIENT_SUBSCRIPTION_INFO *) clnt->subscriptions->get_head ();
01960 while (temp_clnt_info != NULL)
01961 {
01962 if (temp_clnt_info->buffer_number == buffer_number)
01963 {
01964 if (NULL != temp_clnt_info->sub_buf_info)
01965 {
01966 if (NULL != temp_clnt_info->sub_buf_info->sub_clnt_info)
01967 {
01968 temp_clnt_info->sub_buf_info->sub_clnt_info->
01969 delete_node (temp_clnt_info->subscription_list_id);
01970 if (temp_clnt_info->sub_buf_info->sub_clnt_info->
01971 list_size == 0)
01972 {
01973 subscription_buffers->delete_node (temp_clnt_info->
01974 sub_buf_info->
01975 list_id);
01976 delete temp_clnt_info->sub_buf_info->sub_clnt_info;
01977 temp_clnt_info->sub_buf_info->sub_clnt_info = NULL;
01978 delete temp_clnt_info->sub_buf_info;
01979 temp_clnt_info->sub_buf_info = NULL;
01980 }
01981 }
01982 }
01983 delete temp_clnt_info;
01984 temp_clnt_info = NULL;
01985 break;
01986 }
01987 temp_clnt_info =
01988 (TCP_CLIENT_SUBSCRIPTION_INFO *) clnt->subscriptions->get_next ();
01989 }
01990 recalculate_polling_interval ();
01991 }
01992
01993 void
01994 CMS_SERVER_REMOTE_TCP_PORT::recalculate_polling_interval ()
01995 {
01996 int min_poll_interval_millis = 30000;
01997 polling_enabled = 0;
01998 TCP_BUFFER_SUBSCRIPTION_INFO *buf_info =
01999 (TCP_BUFFER_SUBSCRIPTION_INFO *) subscription_buffers->get_head ();
02000 while (NULL != buf_info)
02001 {
02002 TCP_CLIENT_SUBSCRIPTION_INFO *temp_clnt_info =
02003 (TCP_CLIENT_SUBSCRIPTION_INFO *) buf_info->sub_clnt_info->get_head ();
02004 while (temp_clnt_info != NULL)
02005 {
02006 if (temp_clnt_info->poll_interval_millis < min_poll_interval_millis
02007 && temp_clnt_info->subscription_type == CMS_POLLED_SUBSCRIPTION)
02008 {
02009 min_poll_interval_millis = temp_clnt_info->poll_interval_millis;
02010 polling_enabled = 1;
02011 }
02012 temp_clnt_info =
02013 (TCP_CLIENT_SUBSCRIPTION_INFO *)
02014 buf_info->sub_clnt_info->get_next ();
02015 }
02016 buf_info =
02017 (TCP_BUFFER_SUBSCRIPTION_INFO *) subscription_buffers->get_next ();
02018 }
02019 if (min_poll_interval_millis >= ((int) (clk_tck () * 1000.0)))
|
|
||||||||||||
|
Definition at line 2023 of file tcp_srv.cc. 02024 {
02025 current_poll_interval_millis = ((int) (clk_tck () * 1000.0));
02026 }
02027 select_timeout.tv_sec = current_poll_interval_millis / 1000;
02028 select_timeout.tv_usec = (current_poll_interval_millis % 1000) * 1000;
02029 dtimeout = (current_poll_interval_millis + 10) * 1000.0;
02030 if (dtimeout < 0.5)
02031 {
02032 dtimeout = 0.5;
02033 }
02034 }
02035
02036
02037 void
02038 CMS_SERVER_REMOTE_TCP_PORT::update_subscriptions ()
02039 {
02040 #if defined(WIN32) && !defined(gnuwin32)
02041 DWORD pid = GetCurrentProcessId ();
02042 DWORD tid = GetCurrentThreadId ();
02043 #else
02044 #ifdef VXWORKS
02045 int pid = taskIdSelf ();
02046 int tid = 0;
02047 #else
02048 pid_t pid = getpid ();
02049 pid_t tid = 0;
02050 #endif
02051 #endif
02052 CMS_SERVER *server;
02053 server = find_server (pid, tid);
02054 if (NULL == server)
02055 {
02056 rcs_print_error
02057 ("CMS_SERVER_REMOTE_TCP_PORT::update_subscriptions Cannot find server object for pid = %d.\n",
02058 pid);
02059 return;
02060 }
|
|
|
Definition at line 2063 of file tcp_srv.cc. Referenced by add_subscription_client().
02069 {
02070 server->read_req.buffer_number = buf_info->buffer_number;
02071 server->read_req.access_type = CMS_READ_ACCESS;
02072 server->read_req.last_id_read = buf_info->min_last_id;
02073 server->read_reply =
02074 (REMOTE_READ_REPLY *) server->process_request (&server->read_req);
02075 if (NULL == server->read_reply)
02076 {
02077 rcs_print_error ("Server could not process request.\n");
02078 buf_info =
02079 (TCP_BUFFER_SUBSCRIPTION_INFO *)
02080 subscription_buffers->get_next ();
02081 continue;
02082 }
02083 if (server->read_reply->write_id == buf_info->min_last_id ||
02084 server->read_reply->size < 1)
02085 {
02086 buf_info =
02087 (TCP_BUFFER_SUBSCRIPTION_INFO *)
02088 subscription_buffers->get_next ();
02089 continue;
02090 }
02091 *((u_long *) temp_buffer) = 0;
02092 *((u_long *) temp_buffer + 1) = dl_htonl (server->read_reply->status);
02093 *((u_long *) temp_buffer + 2) = dl_htonl (server->read_reply->size);
02094 *((u_long *) temp_buffer + 3) = dl_htonl (server->read_reply->write_id);
02095 *((u_long *) temp_buffer + 4) = dl_htonl (server->read_reply->was_read);
02096 TCP_CLIENT_SUBSCRIPTION_INFO *temp_clnt_info =
02097 (TCP_CLIENT_SUBSCRIPTION_INFO *) buf_info->sub_clnt_info->get_head ();
02098 buf_info->min_last_id = server->read_reply->write_id;
02099 while (temp_clnt_info != NULL)
02100 {
02101 double time_diff = cur_time - temp_clnt_info->last_sub_sent_time;
02102 int time_diff_millis = (int) ((double) time_diff * 1000.0);
02103 rcs_print_debug (PRINT_SERVER_SUBSCRIPTION_ACTIVITY,
|
|
||||||||||||||||||||||||
|
|
|
|
Definition at line 56 of file tcp_srv.hh. |
|
|
Definition at line 58 of file tcp_srv.hh. |
|
|
Definition at line 58 of file tcp_srv.hh. |
|
|
Definition at line 60 of file tcp_srv.hh. |
|
|
Definition at line 61 of file tcp_srv.hh. |
|
|
Definition at line 62 of file tcp_srv.hh. |
|
|
Definition at line 63 of file tcp_srv.hh. |
|
|
Definition at line 64 of file tcp_srv.hh. |
|
|
Definition at line 65 of file tcp_srv.hh. |
|
|
Definition at line 66 of file tcp_srv.hh. |
|
|
Definition at line 67 of file tcp_srv.hh. |
|
|
Definition at line 68 of file tcp_srv.hh. |
|
|
Definition at line 69 of file tcp_srv.hh. |
|
|
Definition at line 70 of file tcp_srv.hh. |
1.2.11.1 written by Dimitri van Heesch,
© 1997-2001